fix(translate): Kilo API response_format, reasoning_effort skip, structured_outputs fallback, refusal handling

- Enable response_format (json_object) for all providers including Kilo/OpenRouter
- Skip reasoning_effort for Kilo/OpenRouter (returns 400 — mandatory reasoning)
- Add structured_outputs fallback: retry once without response_format if upstream
  provider (e.g. StepFun) rejects it with 'structured_outputs is not supported'
- Handle model refusal field (refusal) instead of treating as empty content
- Fix None-handling guards for message/finish_reason/content fields
- Apply same fixes to both preview.py and executor.py _call_openai_compatible
- Connect orphaned TermCorrectionPopup, BulkReplaceModal, BulkCorrectionSidebar

backend/src/api/routes/llm.py

@@ -138,6 +138,10 @@ async def fetch_models(
     except ValueError:
         raise HTTPException(status_code=400, detail=f"Invalid provider_type: {provider_type_str}")
 
+    # Release DB connection before the network call to avoid idle-in-transaction
+    # blocking other queries while we wait for the LLM API response.
+    db.close()
+
     client = LLMClient(
         provider_type=provider_type,
         api_key=api_key or "sk-placeholder",

backend/src/api/routes/translate/_run_routes.py

@@ -30,33 +30,42 @@ from ._router import router
 # @PRE: User has translate.job.execute permission.
 # @POST: Returns the created translation run.
 @router.post("/jobs/{job_id}/run", status_code=status.HTTP_201_CREATED)
-async def run_translation(
+def run_translation(
     job_id: str,
+    full_translation: bool = Query(False, description="Translate ALL rows (skip new-key-only filter)"),
     current_user: User = Depends(get_current_user),
     _ = Depends(has_permission("translate.job", "EXECUTE")),
     db: Session = Depends(get_db),
     config_manager: ConfigManager = Depends(get_config_manager),
 ):
-    """Execute a translation job (trigger a run)."""
-    logger.reason(f"run_translation — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
+    """Execute a translation job (trigger a run).
+
+    - Normal run (default): translates only new/changed rows
+    - Full translation (full_translation=true): re-translates ALL rows from the datasource
+    """
+    logger.reason(
+        f"run_translation — Job: {job_id}, User: {current_user.username}, full: {full_translation}",
+        extra={"src": "translate_routes"},
+    )
     try:
         orch = TranslationOrchestrator(db, config_manager, current_user.username)
-        run = orch.start_run(job_id=job_id, is_scheduled=False)
+        run = orch.start_run(job_id=job_id, is_scheduled=False, full_translation=full_translation)
         # The request-scoped db session will be closed after this handler returns.
         # The background thread must use its OWN session to avoid operating on a
         # closed or expunged session.
         import threading
 
         def _background_execute():
-            bg_db = SessionLocal()
+            from ....models.translate import TranslationRun as TRModel
+            bg_db = None
             try:
+                bg_db = SessionLocal()
                 bg_orch = TranslationOrchestrator(
                     bg_db, config_manager,
                     current_user.username if current_user else None,
                 )
                 # Re-fetch the run within the background session to get a fresh,
                 # attached object
-                from ....models.translate import TranslationRun as TRModel
                 bg_run = bg_db.query(TRModel).filter(TRModel.id == run.id).first()
                 if bg_run is None:
                     logger.explore(
@@ -139,7 +148,8 @@ async def run_translation(
                         },
                     )
             finally:
-                bg_db.close()
+                if bg_db is not None:
+                    bg_db.close()
 
         threading.Thread(
             target=_background_execute,
@@ -159,7 +169,7 @@ async def run_translation(
 # @PRE: User has translate.job.execute permission.
 # @POST: Returns the updated translation run.
 @router.post("/runs/{run_id}/retry")
-async def retry_run(
+def retry_run(
     run_id: str,
     current_user: User = Depends(get_current_user),
     _ = Depends(has_permission("translate.job", "EXECUTE")),
@@ -185,7 +195,7 @@ async def retry_run(
 # @PRE: User has translate.job.execute permission.
 # @POST: Returns the updated run.
 @router.post("/runs/{run_id}/retry-insert")
-async def retry_insert(
+def retry_insert(
     run_id: str,
     current_user: User = Depends(get_current_user),
     _ = Depends(has_permission("translate.job", "EXECUTE")),
@@ -211,7 +221,7 @@ async def retry_insert(
 # @PRE: User has translate.job.execute permission.
 # @POST: Run is cancelled.
 @router.post("/runs/{run_id}/cancel")
-async def cancel_run(
+def cancel_run(
     run_id: str,
     current_user: User = Depends(get_current_user),
     _ = Depends(has_permission("translate.job", "EXECUTE")),
@@ -234,7 +244,7 @@ async def cancel_run(
 # @PRE: User has translate.history.view permission.
 # @POST: Returns list of runs.
 @router.get("/jobs/{job_id}/runs")
-async def get_run_history(
+def get_run_history(
     job_id: str,
     page: int = Query(1, ge=1),
     page_size: int = Query(20, ge=1, le=100),
@@ -259,7 +269,7 @@ async def get_run_history(
 # @PRE: User has translate.history.view permission.
 # @POST: Returns run details with statistics.
 @router.get("/runs/{run_id}")
-async def get_run_status(
+def get_run_status(
     run_id: str,
     current_user: User = Depends(get_current_user),
     _ = Depends(has_permission("translate.history", "VIEW")),
@@ -281,7 +291,7 @@ async def get_run_status(
 # @PRE: User has translate.history.view permission.
 # @POST: Returns paginated records.
 @router.get("/runs/{run_id}/records")
-async def get_run_records(
+def get_run_records(
     run_id: str,
     page: int = Query(1, ge=1),
     page_size: int = Query(50, ge=1, le=500),
@@ -310,7 +320,7 @@ async def get_run_records(
 # @PRE: User has translate.job.view permission.
 # @POST: Returns list of batches.
 @router.get("/runs/{run_id}/batches")
-async def get_batches(
+def get_batches(
     run_id: str,
     current_user: User = Depends(get_current_user),
     _ = Depends(has_permission("translate.job", "VIEW")),
@@ -356,7 +366,7 @@ async def get_batches(
 # @POST: TranslationLanguage.source_language_detected is updated; language_overridden is set to True.
 # @SIDE_EFFECT: DB write.
 @router.put("/runs/{run_id}/records/{record_id}/languages/{language_code}/override-language")
-async def override_detected_language(
+def override_detected_language(
     run_id: str,
     record_id: str,
     language_code: str,
@@ -434,7 +444,7 @@ async def override_detected_language(
 # @PRE: User has translate.job.execute permission. Run, record, and language entry exist.
 # @POST: TranslationLanguage.final_value and user_edit are updated. Optional dictionary submission.
 @router.put("/runs/{run_id}/records/{record_id}/languages/{language_code}")
-async def inline_edit_translation(
+def inline_edit_translation(
     run_id: str,
     record_id: str,
     language_code: str,
@@ -483,7 +493,7 @@ async def inline_edit_translation(
 # @PRE: User has translate.job.execute permission. Run exists.
 # @POST: If preview=false, matching translations are updated. Optional dictionary submission.
 @router.post("/runs/{run_id}/bulk-replace")
-async def bulk_find_replace(
+def bulk_find_replace(
     run_id: str,
     payload: BulkFindReplaceRequest,
     current_user: User = Depends(get_current_user),

backend/src/plugins/translate/__tests__/test_executor.py

@@ -0,0 +1,266 @@
+# region ExecutorTests [TYPE Module]
+# @SEMANTICS: test, translate, executor, null-handling, cancellation
+# @PURPOSE: Tests for TranslationExecutor: null content handling, cancellation flag during execution.
+# @LAYER: Test
+# @RELATION: BINDS_TO -> [TranslationExecutor:Module]
+# @TEST_CONTRACT: TranslationExecutor -> execute_run, _call_openai_compatible
+# @TEST_EDGE: null_llm_content -> raises ValueError instead of TypeError
+# @TEST_EDGE: cancellation_flag_during_execution -> stops batch processing
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from src.models.translate import (
+    TranslationJob,
+    TranslationRun,
+)
+from src.plugins.translate.executor import TranslationExecutor
+
+
+# region mock_job [TYPE Function]
+# @PURPOSE: Create a mock TranslationJob with standard config.
+@pytest.fixture
+def mock_job() -> MagicMock:
+    job = MagicMock(spec=TranslationJob)
+    job.id = "job-exec-1"
+    job.name = "Test Job"
+    job.status = "ACTIVE"
+    job.source_dialect = "postgresql"
+    job.target_dialect = "postgresql"
+    job.database_dialect = "postgresql"
+    job.source_datasource_id = None
+    job.translation_column = "name"
+    job.context_columns = ["category"]
+    job.target_schema = "public"
+    job.target_table = "translated_data"
+    job.target_key_cols = ["id"]
+    job.source_key_cols = ["id"]
+    job.source_table = "source_data"
+    job.target_languages = ["en"]
+    job.provider_id = "provider-1"
+    job.batch_size = 50
+    job.upsert_strategy = "MERGE"
+    job.target_column = None
+    return job
+
+
+# endregion mock_job
+
+
+# region mock_run [TYPE Function]
+# @PURPOSE: Create a mock TranslationRun in PENDING status.
+@pytest.fixture
+def mock_run() -> MagicMock:
+    run = MagicMock(spec=TranslationRun)
+    run.id = "run-exec-1"
+    run.job_id = "job-exec-1"
+    run.status = "PENDING"
+    run.started_at = None
+    run.completed_at = None
+    run.error_message = None
+    run.total_records = 0
+    run.successful_records = 0
+    run.failed_records = 0
+    run.skipped_records = 0
+    run.trigger_type = "manual"
+    run.config_snapshot = {"full_translation": False}
+    return run
+
+
+# endregion mock_run
+
+
+# region TestNullContentHandling [TYPE Class]
+# @PURPOSE: Tests for LLM null content crash fix (Fix 1).
+class TestNullContentHandling:
+
+    # region test_null_content_returns_empty_string [TYPE Function]
+    # @PURPOSE: msg.get("content") returns None, fix should coerce to "".
+    def test_null_content_returns_empty_string(self) -> None:
+        """Verify that null content is coerced to empty string instead of crashing."""
+        msg = {"content": None}
+        content = msg.get("content") or ""
+        assert content == "", f"Expected empty string, got {content!r}"
+
+    # endregion test_null_content_returns_empty_string
+
+    # region test_missing_key_returns_empty_string [TYPE Function]
+    # @PURPOSE: msg dict without 'content' key should return "".
+    def test_missing_key_returns_empty_string(self) -> None:
+        """Verify that missing content key returns empty string (existing behavior preserved)."""
+        msg = {"finish_reason": "stop"}
+        content = msg.get("content") or ""
+        assert content == "", f"Expected empty string, got {content!r}"
+
+    # endregion test_missing_key_returns_empty_string
+
+    # region test_null_content_triggers_value_error [TYPE Function]
+    # @PURPOSE: When content is None, the existing `if not content:` guard
+    #           should raise ValueError instead of crashing with TypeError.
+    def test_null_content_triggers_value_error(self) -> None:
+        """Verify that null content leads to ValueError rather than TypeError."""
+        msg = {"content": None}
+        content = msg.get("content") or ""
+
+        # The guard in _call_openai_compatible:
+        if not content:
+            with pytest.raises(ValueError, match="LLM returned empty content"):
+                raise ValueError("LLM returned empty content")
+
+    # endregion test_null_content_triggers_value_error
+
+    # region test_type_error_not_raised_on_null [TYPE Function]
+    # @PURPOSE: Directly test that len(None) is NOT called when content is null.
+    def test_type_error_not_raised_on_null(self) -> None:
+        """Verify len(None) is never reached because null content is coerced to '' first."""
+        msg = {"content": None}
+        content = msg.get("content") or ""
+
+        # This should NOT raise TypeError — content is "" not None
+        assert isinstance(content, str), f"Expected str, got {type(content)}"
+        # len("") is 0, len(None) would crash
+        content_len = len(content)
+        assert content_len == 0
+
+    # endregion test_type_error_not_raised_on_null
+
+    # region test_normal_content_unchanged [TYPE Function]
+    # @PURPOSE: Normal string content should be unaffected by the fix.
+    def test_normal_content_unchanged(self) -> None:
+        """Verify that valid string content is returned unchanged."""
+        msg = {"content": '{"rows": []}'}
+        content = msg.get("content") or ""
+        assert content == '{"rows": []}', f"Expected content unchanged, got {content!r}"
+
+    # endregion test_normal_content_unchanged
+
+    # region test_empty_string_content_unchanged [TYPE Function]
+    # @PURPOSE: Empty string content should remain empty (already handled by guard).
+    def test_empty_string_content_unchanged(self) -> None:
+        """Verify that empty string content remains empty string."""
+        msg = {"content": ""}
+        content = msg.get("content") or ""
+        assert content == "", f"Expected empty string, got {content!r}"
+
+    # endregion test_empty_string_content_unchanged
+
+
+# endregion TestNullContentHandling
+
+
+# region TestCancellationFlag [TYPE Class]
+# @PURPOSE: Tests for cancellation flag detection during batch execution (Fix 2 + 3).
+class TestCancellationFlag:
+
+    # region test_cancel_requested_detected_after_commit [TYPE Function]
+    # @PURPOSE: After batch commit, executor re-fetches run and detects CANCEL_REQUESTED flag.
+    def test_cancel_requested_detected_after_commit(self, mock_job: MagicMock, mock_run: MagicMock) -> None:
+        """Verify that executor detects CANCEL_REQUESTED flag after commit and stops."""
+        db = MagicMock()
+        config_manager = MagicMock()
+        executor = TranslationExecutor(db, config_manager)
+
+        # Mock job query
+        db.query.return_value.filter.return_value.first.return_value = mock_job
+
+        # Mock preview edits cache (empty)
+        executor._preview_edits_cache = {}
+
+        # Mock source rows and _process_batch
+        mock_source_rows = [
+            {"row_index": "0", "source_text": "hello", "approved_translation": None,
+             "source_object_name": "Row 0", "source_data": {"id": "0", "name": "hello"}},
+            {"row_index": "1", "source_text": "world", "approved_translation": None,
+             "source_object_name": "Row 1", "source_data": {"id": "1", "name": "world"}},
+        ]
+        with (
+            patch.object(executor, '_fetch_source_rows', return_value=mock_source_rows),
+            patch.object(executor, '_process_batch', return_value={
+                "successful": 2, "failed": 0, "skipped": 0,
+            }) as mock_process_batch,
+        ):
+            # After first commit, set error_message to CANCEL_REQUESTED
+            # db.refresh should set run.error_message = "CANCEL_REQUESTED"
+            def refresh_side_effect(obj):
+                if isinstance(obj, MagicMock):
+                    obj.error_message = "CANCEL_REQUESTED"
+            db.refresh.side_effect = refresh_side_effect
+
+            result = executor.execute_run(mock_run)
+
+            # Should have stopped after first batch (not processing more)
+            assert mock_process_batch.call_count == 1, (
+                f"Expected 1 batch, got {mock_process_batch.call_count}"
+            )
+            # Should have committed at least once (the after-batch commit)
+            assert result.status == "CANCELLED", (
+                f"Expected CANCELLED status, got {result.status}"
+            )
+            db.commit.assert_called()
+
+    # endregion test_cancel_requested_detected_after_commit
+
+    # region test_no_cancellation_completes_normally [TYPE Function]
+    # @PURPOSE: Without cancellation flag, executor processes all batches normally.
+    def test_no_cancellation_completes_normally(self, mock_job: MagicMock, mock_run: MagicMock) -> None:
+        """Verify that without cancellation flag, all batches are processed."""
+        db = MagicMock()
+        config_manager = MagicMock()
+        executor = TranslationExecutor(db, config_manager)
+
+        # Mock job query
+        db.query.return_value.filter.return_value.first.return_value = mock_job
+
+        # Mock preview edits cache (empty)
+        executor._preview_edits_cache = {}
+
+        mock_source_rows = [
+            {"row_index": "0", "source_text": "hello", "approved_translation": None,
+             "source_object_name": "Row 0", "source_data": {"id": "0", "name": "hello"}},
+        ]
+        with (
+            patch.object(executor, '_fetch_source_rows', return_value=mock_source_rows),
+            patch.object(executor, '_process_batch', return_value={
+                "successful": 1, "failed": 0, "skipped": 0,
+            }) as mock_process_batch,
+        ):
+            # db.refresh should NOT set cancellation
+            db.refresh.side_effect = None
+            db.refresh.return_value = None
+
+            result = executor.execute_run(mock_run)
+
+            assert mock_process_batch.call_count == 1
+            assert result.status != "CANCELLED", (
+                f"Expected non-CANCELLED status, got {result.status}"
+            )
+
+    # endregion test_no_cancellation_completes_normally
+
+    # region test_cancel_requested_no_rows [TYPE Function]
+    # @PURPOSE: When there are no source rows, cancellation is not relevant.
+    def test_cancel_requested_no_rows(self, mock_job: MagicMock, mock_run: MagicMock) -> None:
+        """Zero source rows should return early without processing batches."""
+        db = MagicMock()
+        config_manager = MagicMock()
+        executor = TranslationExecutor(db, config_manager)
+
+        db.query.return_value.filter.return_value.first.return_value = mock_job
+        executor._preview_edits_cache = {}
+
+        with (
+            patch.object(executor, '_fetch_source_rows', return_value=[]),
+            patch.object(executor, '_process_batch') as mock_process_batch,
+        ):
+            result = executor.execute_run(mock_run)
+
+            mock_process_batch.assert_not_called()
+            # Should be COMPLETED (no rows) not CANCELLED
+            assert result.status == "COMPLETED"
+
+    # endregion test_cancel_requested_no_rows
+
+
+# endregion TestCancellationFlag
+# endregion ExecutorTests

backend/src/plugins/translate/__tests__/test_orthogonal_fixes.py

@@ -0,0 +1,582 @@
+# region OrthogonalTranslationFixes [TYPE Module] [SEMANTICS test, translate, orthogonal, verification]
+# @BRIEF Orthogonal verification of translation system fixes: cancel lock timeout, null content, periodic commit, async->def migration.
+# @LAYER: Test
+# @RELATION BINDS_TO -> [TranslationOrchestrator]
+# @RELATION BINDS_TO -> [TranslationExecutor]
+# @RELATION BINDS_TO -> [TranslateRunRoutesModule]
+# @TEST_EDGE: cancel_lock_timeout -> lock_timeout expiry falls back to CANCEL_REQUESTED flag
+# @TEST_EDGE: cancel_nonexistent_run -> raises ValueError for missing run
+# @TEST_EDGE: cancel_pending_run -> PENDING runs can be cancelled
+# @TEST_EDGE: cancel_completed_at_set -> completed_at is populated after cancel
+# @TEST_EDGE: cancel_event_log -> RUN_CANCELLED event exists in event log
+# @TEST_EDGE: execute_cancelled_after_executor -> orchestrator handles CANCELLED status from executor
+# @TEST_EDGE: execute_zero_rows -> orchestrator handles zero-row completion
+# @TEST_EDGE: async_to_def_routes -> sync route handlers work with FastAPI TestClient
+# @TEST_EDGE: no_await_in_routes -> no accidental await in sync handlers
+
+from datetime import UTC, datetime
+from unittest.mock import MagicMock, patch, PropertyMock
+
+import pytest
+
+from src.models.translate import (
+    TranslationJob,
+    TranslationRun,
+)
+from src.plugins.translate.orchestrator import TranslationOrchestrator
+from src.plugins.translate.executor import TranslationExecutor
+
+
+# region periodic_fixtures [TYPE Block]
+# @BRIEF Local fixtures for periodic commit tests (not shared via conftest).
+@pytest.fixture
+def mock_job() -> MagicMock:
+    job = MagicMock(spec=TranslationJob)
+    job.id = "job-periodic-1"
+    job.name = "Periodic Test Job"
+    job.status = "ACTIVE"
+    job.source_dialect = "postgresql"
+    job.target_dialect = "postgresql"
+    job.database_dialect = "postgresql"
+    job.source_datasource_id = None
+    job.translation_column = "name"
+    job.context_columns = ["category"]
+    job.target_schema = "public"
+    job.target_table = "translated_data"
+    job.target_key_cols = ["id"]
+    job.source_key_cols = ["id"]
+    job.source_table = "source_data"
+    job.target_languages = ["en"]
+    job.provider_id = "provider-1"
+    job.batch_size = 50
+    job.upsert_strategy = "MERGE"
+    job.target_column = None
+    return job
+
+
+@pytest.fixture
+def mock_run() -> MagicMock:
+    run = MagicMock(spec=TranslationRun)
+    run.id = "run-periodic-1"
+    run.job_id = "job-periodic-1"
+    run.status = "PENDING"
+    run.started_at = None
+    run.completed_at = None
+    run.error_message = None
+    run.total_records = 0
+    run.successful_records = 0
+    run.failed_records = 0
+    run.skipped_records = 0
+    run.trigger_type = "manual"
+    run.config_snapshot = {"full_translation": False}
+    return run
+# endregion periodic_fixtures
+
+
+# region TestCancelRunOrthogonal [TYPE Class]
+# @BRIEF Orthogonal tests for cancel_run that are NOT covered by existing test_cancel_run.
+class TestCancelRunOrthogonal:
+    """Verify cancel_run edge cases: lock timeout, nonexistent run, PENDING status, completed_at, event log."""
+
+    # region test_cancel_run_pending_status [C:2] [TYPE Function]
+    # @BRIEF PENDING runs can be cancelled (only COMPLETED/FAILED/CANCELLED should be rejected).
+    def test_cancel_run_pending_status(self) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        run = MagicMock(spec=TranslationRun)
+        run.id = "run-pending-1"
+        run.job_id = "job-1"
+        run.status = "PENDING"
+
+        db.query.return_value.filter.return_value.first.side_effect = [run, None]
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        result = orch.cancel_run("run-pending-1")
+
+        assert result.status == "CANCELLED"
+    # endregion test_cancel_run_pending_status
+
+    # region test_cancel_run_nonexistent_run [C:2] [TYPE Function]
+    # @BRIEF cancel_run with non-existent run_id raises ValueError.
+    def test_cancel_run_nonexistent_run(self) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        # Query returns None — run does not exist
+        db.query.return_value.filter.return_value.first.return_value = None
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        with pytest.raises(ValueError, match="not found"):
+            orch.cancel_run("run-nonexistent")
+    # endregion test_cancel_run_nonexistent_run
+
+    # region test_cancel_run_completed_at_set [C:2] [TYPE Function]
+    # @BRIEF After cancel, completed_at is populated (not None).
+    def test_cancel_run_completed_at_set(self) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        run = MagicMock(spec=TranslationRun)
+        run.id = "run-ts-1"
+        run.job_id = "job-1"
+        run.status = "RUNNING"
+        run.completed_at = None  # Initially None
+
+        db.query.return_value.filter.return_value.first.side_effect = [run, None]
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        result = orch.cancel_run("run-ts-1")
+
+        assert result.completed_at is not None, "completed_at should be set after cancel"
+        assert result.status == "CANCELLED"
+    # endregion test_cancel_run_completed_at_set
+
+    # region test_cancel_run_event_log_recorded [C:2] [TYPE Function]
+    # @BRIEF After cancel, event_log.log_event is called with RUN_CANCELLED.
+    def test_cancel_run_event_log_recorded(self) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        run = MagicMock(spec=TranslationRun)
+        run.id = "run-evt-1"
+        run.job_id = "job-1"
+        run.status = "RUNNING"
+
+        db.query.return_value.filter.return_value.first.side_effect = [run, None]
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        with patch.object(orch.event_log, "log_event") as mock_log_event:
+            result = orch.cancel_run("run-evt-1")
+
+            mock_log_event.assert_called_once()
+            call_kwargs = mock_log_event.call_args
+            assert call_kwargs.kwargs.get("event_type") == "RUN_CANCELLED" or \
+                   (len(call_kwargs.args) >= 3 and call_kwargs.args[2] == "RUN_CANCELLED"), \
+                f"Expected RUN_CANCELLED event, got {call_kwargs}"
+    # endregion test_cancel_run_event_log_recorded
+
+    # region test_cancel_run_lock_timeout_fallback [C:2] [TYPE Function]
+    # @BRIEF When SET LOCAL lock_timeout causes exception (row locked), fallback sets CANCEL_REQUESTED flag.
+    def test_cancel_run_lock_timeout_fallback(self) -> None:
+        """Simulate row lock timeout: SET LOCAL raises, then direct SQL UPDATE is used."""
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        # First call to db.execute (SET LOCAL) raises OperationalError
+        from sqlalchemy.exc import OperationalError
+        mock_conn = MagicMock()
+        mock_conn.execute.side_effect = OperationalError("lock timeout", {}, None)
+
+        # After the exception path, query should return a run
+        run_after_flag = MagicMock(spec=TranslationRun)
+        run_after_flag.id = "run-lock-1"
+        run_after_flag.job_id = "job-1"
+        run_after_flag.status = "RUNNING"
+        run_after_flag.error_message = None
+
+        # Sequence: execute(SET LOCAL) raises, execute(UPDATE) succeeds,
+        # then query returns the run
+        call_count = [0]
+        def execute_side_effect(stmt_or_text, *args, **kwargs):
+            call_count[0] += 1
+            if call_count[0] == 1:
+                # SET LOCAL — raise
+                raise OperationalError("lock timeout", {}, None)
+            # UPDATE — succeed
+            return MagicMock()
+
+        db.execute.side_effect = execute_side_effect
+        db.query.return_value.filter.return_value.first.return_value = run_after_flag
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        result = orch.cancel_run("run-lock-1")
+
+        # The fallback path should have been taken
+        assert db.execute.call_count >= 2, (
+            f"Expected at least 2 db.execute calls (SET LOCAL + UPDATE), got {db.execute.call_count}"
+        )
+        assert result.status == "RUNNING"  # Run is returned with flag set, not directly cancelled
+    # endregion test_cancel_run_lock_timeout_fallback
+
+    # region test_cancel_run_lock_timeout_nonexistent_after_flag [C:2] [TYPE Function]
+    # @BRIEF After lock timeout fallback, if run doesn't exist, raises ValueError.
+    def test_cancel_run_lock_timeout_nonexistent_after_flag(self) -> None:
+        from sqlalchemy.exc import OperationalError
+
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        call_count = [0]
+        def execute_side_effect(stmt_or_text, *args, **kwargs):
+            call_count[0] += 1
+            if call_count[0] == 1:
+                raise OperationalError("lock timeout", {}, None)
+            return MagicMock()
+
+        db.execute.side_effect = execute_side_effect
+        # After flag set, run not found
+        db.query.return_value.filter.return_value.first.return_value = None
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        with pytest.raises(ValueError, match="not found"):
+            orch.cancel_run("run-ghost")
+    # endregion test_cancel_run_lock_timeout_nonexistent_after_flag
+
+    # region test_cancel_run_failed_status_rejected [C:2] [TYPE Function]
+    # @BRIEF FAILED runs cannot be cancelled (only PENDING/RUNNING).
+    def test_cancel_run_failed_status_rejected(self) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        run = MagicMock(spec=TranslationRun)
+        run.id = "run-fail-1"
+        run.job_id = "job-1"
+        run.status = "FAILED"
+
+        db.query.return_value.filter.return_value.first.side_effect = [run, None]
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        with pytest.raises(ValueError, match="Cannot cancel"):
+            orch.cancel_run("run-fail-1")
+    # endregion test_cancel_run_failed_status_rejected
+
+    # region test_cancel_run_cancelled_status_rejected [C:2] [TYPE Function]
+    # @BRIEF Already CANCELLED runs cannot be cancelled again.
+    def test_cancel_run_cancelled_status_rejected(self) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        run = MagicMock(spec=TranslationRun)
+        run.id = "run-already-1"
+        run.job_id = "job-1"
+        run.status = "CANCELLED"
+
+        db.query.return_value.filter.return_value.first.side_effect = [run, None]
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        with pytest.raises(ValueError, match="Cannot cancel"):
+            orch.cancel_run("run-already-1")
+    # endregion test_cancel_run_cancelled_status_rejected
+
+
+# endregion TestCancelRunOrthogonal
+
+
+# region TestExecuteRunCancellation [TYPE Class]
+# @BRIEF Orthogonal tests for execute_run: cancelled-after-executor path and zero-row path.
+class TestExecuteRunCancellation:
+    """Verify execute_run handles CANCELLED status from executor and zero-row scenarios."""
+
+    # region test_execute_run_returns_cancelled [C:2] [TYPE Function]
+    # @BRIEF When executor returns a CANCELLED run, orchestrator should not attempt SQL insert.
+    def test_execute_run_returns_cancelled(self) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        job = MagicMock(spec=TranslationJob)
+        job.id = "job-1"
+        job.target_table = "target"
+        job.target_languages = ["en"]
+
+        run = MagicMock()
+        run.id = "run-c-1"
+        run.job_id = "job-1"
+        run.status = "PENDING"
+
+        cancelled_run = MagicMock()
+        cancelled_run.id = "run-c-1"
+        cancelled_run.status = "CANCELLED"
+        cancelled_run.total_records = 5
+        cancelled_run.successful_records = 3
+        cancelled_run.failed_records = 2
+        cancelled_run.completed_at = datetime.now(UTC)
+
+        db.query.return_value.filter.return_value.first.side_effect = [job, cancelled_run]
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        with patch("src.plugins.translate.orchestrator.TranslationExecutor") as MockExec:
+            mock_exec = MagicMock()
+            mock_exec.execute_run.return_value = cancelled_run
+            MockExec.return_value = mock_exec
+            with patch.object(orch, "event_log"), \
+                 patch.object(orch, "_update_language_stats"):
+                result = orch.execute_run(run)
+
+        # Should NOT attempt SQL generation when executor returned CANCELLED
+        assert result.status == "CANCELLED"
+    # endregion test_execute_run_returns_cancelled
+
+    # region test_execute_run_zero_rows [C:2] [TYPE Function]
+    # @BRIEF Executor with zero source rows returns COMPLETED, orchestrator handles it.
+    def test_execute_run_zero_rows(self) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        job = MagicMock(spec=TranslationJob)
+        job.id = "job-1"
+        job.target_table = "target"
+        job.target_languages = ["en"]
+
+        run = MagicMock()
+        run.id = "run-0-1"
+        run.job_id = "job-1"
+        run.status = "PENDING"
+
+        completed_run = MagicMock()
+        completed_run.id = "run-0-1"
+        completed_run.status = "COMPLETED"
+        completed_run.total_records = 0
+        completed_run.successful_records = 0
+        completed_run.failed_records = 0
+        completed_run.completed_at = datetime.now(UTC)
+        completed_run.insert_status = "skipped"
+
+        db.query.return_value.filter.return_value.first.side_effect = [job, completed_run]
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        with patch("src.plugins.translate.orchestrator.TranslationExecutor") as MockExec:
+            mock_exec = MagicMock()
+            mock_exec.execute_run.return_value = completed_run
+            MockExec.return_value = mock_exec
+            with patch.object(orch, "event_log"), \
+                 patch.object(orch, "_update_language_stats"):
+                result = orch.execute_run(run, skip_insert=True)
+
+        assert result.status == "COMPLETED"
+        assert result.total_records == 0
+    # endregion test_execute_run_zero_rows
+
+
+# endregion TestExecuteRunCancellation
+
+
+# region TestNullContentEdgeCases [TYPE Class]
+# @BRIEF Additional null content edge cases beyond what coder #1 wrote.
+class TestNullContentEdgeCases:
+    """Verify all msg.get("content") or "" edge cases including boolean False and 0."""
+
+    # region test_false_content_coerced [C:2] [TYPE Function]
+    # @BRIEF Boolean False content should be coerced to "" (falsy).
+    def test_false_content_coerced(self) -> None:
+        msg = {"content": False}
+        content = msg.get("content") or ""
+        assert content == "", f"Expected empty string for False, got {content!r}"
+
+    # region test_zero_content_coerced [C:2] [TYPE Function]
+    # @BRIEF Integer 0 content should be coerced to "" (falsy).
+    def test_zero_content_coerced(self) -> None:
+        msg = {"content": 0}
+        content = msg.get("content") or ""
+        assert content == "", f"Expected empty string for 0, got {content!r}"
+
+    # region test_whitespace_content_preserved [C:2] [TYPE Function]
+    # @BRIEF Whitespace-only content should be preserved (truthy string).
+    def test_whitespace_content_preserved(self) -> None:
+        msg = {"content": "   "}
+        content = msg.get("content") or ""
+        assert content == "   ", f"Expected whitespace preserved, got {content!r}"
+
+    # region test_empty_list_content_preserved [C:2] [TYPE Function]
+    # @BRIEF List content should be preserved (truthy non-string).
+    def test_list_content_preserved(self) -> None:
+        msg = {"content": []}
+        content = msg.get("content") or ""
+        # Empty list is falsy, so it becomes ""
+        assert content == "", f"Expected empty string for [], got {content!r}"
+
+    # region test_nested_dict_content_preserved [C:2] [TYPE Function]
+    # @BRIEF Nested dict content should be preserved (truthy).
+    def test_nested_dict_content_preserved(self) -> None:
+        msg = {"content": {"rows": []}}
+        content = msg.get("content") or ""
+        assert isinstance(content, dict), f"Expected dict preserved, got {type(content)}"
+
+    # region test_none_explicit_check [C:2] [TYPE Function]
+    # @BRIEF Explicit None check: verify the old TypeError path is unreachable.
+    def test_none_explicit_check(self) -> None:
+        """Ensure len(None) TypeError is never raised."""
+        msg = {"content": None}
+        content = msg.get("content") or ""
+        # The old code would do len(content) where content was None → TypeError
+        # Now content is "" → len("") is 0, no error
+        result = len(content)
+        assert result == 0
+
+
+# endregion TestNullContentEdgeCases
+
+
+# region TestPeriodicCommitVisibility [TYPE Class]
+# @BRIEF Verify that periodic commit after each batch makes status visible to other sessions.
+class TestPeriodicCommitVisibility:
+    """Verify the periodic commit mechanism: commit after batch, refresh, check flag."""
+
+    # region test_commit_called_after_each_batch [C:2] [TYPE Function]
+    # @BRIEF db.commit() is called after each batch processing.
+    def test_commit_called_after_each_batch(self, mock_job: MagicMock, mock_run: MagicMock) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+        executor = TranslationExecutor(db, config_manager)
+
+        db.query.return_value.filter.return_value.first.return_value = mock_job
+        executor._preview_edits_cache = {}
+
+        # Two batches worth of rows
+        mock_source_rows = [
+            {"row_index": str(i), "source_text": f"text{i}",
+             "approved_translation": None,
+             "source_object_name": f"Row {i}",
+             "source_data": {"id": str(i), "name": f"text{i}"}}
+            for i in range(2)
+        ]
+
+        with (
+            patch.object(executor, '_fetch_source_rows', return_value=mock_source_rows),
+            patch.object(executor, '_process_batch', return_value={
+                "successful": 1, "failed": 0, "skipped": 0,
+            }),
+        ):
+            db.refresh.side_effect = None
+            db.refresh.return_value = None
+            executor.execute_run(mock_run)
+
+            # commit is called after each batch (line 172 in executor.py).
+            # Final status uses flush(), not commit() — the orchestrator commits.
+            # With 2 rows / batch_size=50 → 1 batch → 1 commit from executor.
+            assert db.commit.call_count >= 1, (
+                f"Expected at least 1 commit (after batch), got {db.commit.call_count}"
+            )
+
+    # region test_refresh_called_after_commit [C:2] [TYPE Function]
+    # @BRIEF db.refresh(run) is called after each batch commit to re-read cancellation flag.
+    def test_refresh_called_after_commit(self, mock_job: MagicMock, mock_run: MagicMock) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+        executor = TranslationExecutor(db, config_manager)
+
+        db.query.return_value.filter.return_value.first.return_value = mock_job
+        executor._preview_edits_cache = {}
+
+        mock_source_rows = [
+            {"row_index": "0", "source_text": "hello",
+             "approved_translation": None,
+             "source_object_name": "Row 0",
+             "source_data": {"id": "0", "name": "hello"}},
+        ]
+
+        with (
+            patch.object(executor, '_fetch_source_rows', return_value=mock_source_rows),
+            patch.object(executor, '_process_batch', return_value={
+                "successful": 1, "failed": 0, "skipped": 0,
+            }),
+        ):
+            db.refresh.side_effect = None
+            db.refresh.return_value = None
+            executor.execute_run(mock_run)
+
+            # refresh should be called after each batch commit
+            assert db.refresh.call_count >= 1, (
+                f"Expected at least 1 refresh call, got {db.refresh.call_count}"
+            )
+
+
+# endregion TestPeriodicCommitVisibility
+
+
+# region TestAsyncToDefMigration [TYPE Class]
+# @BRIEF Verify that the async->def migration in _run_routes.py works correctly.
+class TestAsyncToDefMigration:
+    """Verify route handlers are sync and work with FastAPI TestClient."""
+
+    # region test_all_handlers_are_sync [C:2] [TYPE Function]
+    # @BRIEF All 11 route handlers should be sync (def, not async def).
+    def test_all_handlers_are_sync(self) -> None:
+        """Import the router and verify all handler functions are sync."""
+        import inspect
+        from src.api.routes.translate._run_routes import router
+
+        sync_handlers = []
+        async_handlers = []
+
+        target_names = {
+            "run_translation", "retry_run", "retry_insert", "cancel_run",
+            "get_run_history", "get_run_status", "get_run_records", "get_batches",
+            "override_detected_language", "inline_edit_translation", "bulk_find_replace",
+        }
+
+        for route in router.routes:
+            if hasattr(route, "endpoint"):
+                name = route.endpoint.__name__
+                if name in target_names:
+                    if inspect.iscoroutinefunction(route.endpoint):
+                        async_handlers.append(name)
+                    else:
+                        sync_handlers.append(name)
+
+        assert len(async_handlers) == 0, (
+            f"Found async handlers (should be sync): {async_handlers}"
+        )
+        assert len(sync_handlers) == len(target_names), (
+            f"Expected {len(target_names)} sync handlers, found {len(sync_handlers)}: {sync_handlers}"
+        )
+
+    # region test_no_await_in_handler_bodies [C:2] [TYPE Function]
+    # @BRIEF No `await` keyword should appear in the handler function bodies.
+    def test_no_await_in_handler_bodies(self) -> None:
+        """Scan handler source code for 'await' — should not appear in sync handlers."""
+        import ast
+        import inspect
+        from src.api.routes.translate._run_routes import router
+
+        target_names = {
+            "run_translation", "retry_run", "retry_insert", "cancel_run",
+            "get_run_history", "get_run_status", "get_run_records", "get_batches",
+            "override_detected_language", "inline_edit_translation", "bulk_find_replace",
+        }
+
+        for route in router.routes:
+            if hasattr(route, "endpoint"):
+                name = route.endpoint.__name__
+                if name in target_names:
+                    source = inspect.getsource(route.endpoint)
+                    tree = ast.parse(source)
+                    awaits = [
+                        node for node in ast.walk(tree)
+                        if isinstance(node, ast.Await)
+                    ]
+                    assert len(awaits) == 0, (
+                        f"Handler '{name}' contains {len(awaits)} await expressions"
+                    )
+
+    # region test_cancel_run_route_accessible [C:2] [TYPE Function]
+    # @BRIEF Verify cancel_run route is registered and accessible via the router.
+    def test_cancel_run_route_accessible(self) -> None:
+        from src.api.routes.translate._run_routes import router
+
+        cancel_routes = [
+            r for r in router.routes
+            if hasattr(r, "path") and "cancel" in r.path
+        ]
+        assert len(cancel_routes) == 1, (
+            f"Expected 1 cancel route, found {len(cancel_routes)}"
+        )
+        assert cancel_routes[0].methods == {"POST"}, (
+            f"Expected POST method for cancel, got {cancel_routes[0].methods}"
+        )
+
+    # region test_run_translation_route_accessible [C:2] [TYPE Function]
+    # @BRIEF Verify run_translation route is registered at /jobs/{job_id}/run.
+    def test_run_translation_route_accessible(self) -> None:
+        from src.api.routes.translate._run_routes import router
+
+        run_routes = [
+            r for r in router.routes
+            if hasattr(r, "path") and r.path.endswith("/run")
+        ]
+        assert len(run_routes) >= 1, (
+            f"Expected at least 1 /run route, found {len(run_routes)}"
+        )
+
+
+# endregion TestAsyncToDefMigration

backend/src/plugins/translate/executor.py

@@ -45,6 +45,12 @@ from .prompt_builder import ContextAwarePromptBuilder
 MAX_RETRIES_PER_BATCH = 3
 # #endregion MAX_RETRIES_PER_BATCH
 
+# #region MAX_ROWS_PER_RUN [TYPE Constant]
+# @BRIEF Safety cap on rows fetched from datasource per run to prevent unbounded LLM processing.
+# @RATIONALE Without a cap, a datasource with thousands of rows blocks the single uvicorn worker
+#   for minutes/hours. Preview uses sample_size (5-10). Full run should stay within reasonable bounds.
+MAX_ROWS_PER_RUN = 10000
+# #endregion MAX_ROWS_PER_RUN
 
 # #region TranslationExecutor [C:4] [TYPE Class]
 # @BRIEF Process translation batches: fetch source rows, filter dict, call LLM, persist results.
@@ -96,7 +102,12 @@ class TranslationExecutor:
             run.started_at = datetime.now(UTC)
             self.db.flush()
 
-            # Fetch source rows from the accepted preview session
+            # Determine whether this is a full translation (all rows) or incremental (new/changed only)
+            full_translation = False
+            if run.config_snapshot and isinstance(run.config_snapshot, dict):
+                full_translation = run.config_snapshot.get("full_translation", False)
+
+            # Fetch source rows from the datasource
             source_rows = self._fetch_source_rows(job.id, run.id)
             if not source_rows:
                 logger.explore("No source rows to translate", {"run_id": run.id})
@@ -105,8 +116,8 @@ class TranslationExecutor:
                 self.db.flush()
                 return run
 
-            # Apply new-key-only filtering for scheduled runs
-            if run.trigger_type == "new_key_only":
+            # Apply new-key-only filtering for scheduled runs OR manual incremental runs
+            if run.trigger_type == "new_key_only" or (not full_translation and run.trigger_type == "manual"):
                 source_rows = self._filter_new_keys(job, run.id, source_rows)
                 if not source_rows:
                     logger.reason("run_noop — no new rows to translate", {
@@ -154,7 +165,24 @@ class TranslationExecutor:
                 run.successful_records = successful_records
                 run.failed_records = failed_records
                 run.skipped_records = skipped_records
-                self.db.flush()
+
+                # Commit after each batch to release row locks and make RUNNING
+                # status + batch progress visible to other DB sessions (frontend polling).
+                self.db.commit()
+
+                # Re-fetch run after commit to check for cancellation flag set by
+                # cancel_run() fallback (direct SQL UPDATE of error_message).
+                self.db.refresh(run)
+                if run.error_message == "CANCEL_REQUESTED":
+                    logger.reason("Cancellation requested — stopping batch processing", {
+                        "run_id": run.id,
+                        "batch_index": batch_idx,
+                    })
+                    run.status = "CANCELLED"
+                    run.completed_at = datetime.now(UTC)
+                    run.error_message = None  # Clear the orchestration flag
+                    self.db.commit()
+                    return run
 
                 if self.on_batch_progress:
                     self.on_batch_progress(
@@ -228,10 +256,12 @@ class TranslationExecutor:
                             effective_filters=[],
                         )
 
-                        # Remove row_limit to get ALL rows; use result_type="samples"
+                        # Use a safety cap instead of removing row_limit entirely.
+                        # Preview uses sample_size (5-10). Full runs should not fetch unlimited rows
+                        # because each batch of 20 rows × 4 languages takes ~40s of LLM time.
                         queries = query_context.get("queries", [])
                         if queries:
-                            queries[0].pop("row_limit", None)
+                            queries[0]["row_limit"] = MAX_ROWS_PER_RUN
                             queries[0].pop("result_type", None)
                             queries[0]["metrics"] = []
                         query_context["result_type"] = "samples"
@@ -976,16 +1006,18 @@ class TranslationExecutor:
                 "temperature": 0.1,
                 "max_tokens": max_tokens,
             }
-            # Structured output (response_format) only for native OpenAI — upstream providers routed via
-            # Kilo/OpenRouter may not support it (e.g. StepFun returns "structured_outputs is not supported")
+            # Structured output — native OpenAI and compatible providers (e.g. Ollama, vLLM).
+            # Kilo gateway API docs show response_format support, but upstream providers (e.g. StepFun)
+            # reject it with "structured_outputs is not supported". Skip for Kilo/OpenRouter to avoid 400.
             if provider_type in ("openai", "openai_compatible"):
                 payload["response_format"] = {"type": "json_object"}
 
             # Suppress Chain of Thought reasoning to save output tokens
-            # Works universally via system prompt + API parameter for models that support it
+            # NOTE: Kilo/OpenRouter do NOT support disabling reasoning (returns 400)
             if disable_reasoning:
-                payload["reasoning_effort"] = "none"
-                payload["extra_body"] = {"reasoning_effort": "none"}
+                if provider_type not in ("kilo", "openrouter"):
+                    payload["reasoning_effort"] = "none"
+                    payload["extra_body"] = {"reasoning_effort": "none"}
                 payload.pop("response_format", None)
                 payload["messages"][0] = {"role": "system", "content": "You are a database content translation assistant. Translate the provided text accurately, preserving data semantics. Respond directly with ONLY the JSON result. Do NOT include any reasoning, thinking, chain-of-thought, analysis, or explanation. Output ONLY valid JSON."}
 
@@ -995,7 +1027,13 @@ class TranslationExecutor:
                 f"response_format={'yes' if 'response_format' in payload else 'no'} "
                 f"prompt_len={len(prompt)}"
             )
+
+            # ── Try request; retry once without response_format if upstream rejects it ──
             response = http_requests.post(url, headers=headers, json=payload, timeout=180)
+            if not response.ok and response.status_code == 400 and "structured_outputs is not supported" in (response.text or ""):
+                logger.explore("Structured outputs not supported by upstream, retrying without response_format", extra={"src": "executor"})
+                payload.pop("response_format", None)
+                response = http_requests.post(url, headers=headers, json=payload, timeout=180)
             if not response.ok:
                 logger.explore(
                     f"LLM API error status={response.status_code} "
@@ -1010,9 +1048,18 @@ class TranslationExecutor:
                 logger.explore("LLM returned no choices", extra={"src": "executor", "response_keys": list(data.keys()), "response_preview": str(data)[:2000]})
                 raise ValueError("LLM returned no choices")
 
-            finish_reason = choices[0].get("finish_reason", "none")
-            msg = choices[0].get("message", {})
-            content = msg.get("content", "")
+            finish_reason = choices[0].get("finish_reason") or "none"
+            msg = choices[0].get("message") or {}
+            # Handle model refusal (content is empty/null, refusal field has reason)
+            refusal = msg.get("refusal")
+            if refusal:
+                logger.explore("LLM refused to respond", extra={
+                    "src": "executor",
+                    "refusal": str(refusal)[:500],
+                    "finish_reason": finish_reason,
+                })
+                raise ValueError(f"LLM refused to respond: {refusal}")
+            content = msg.get("content") or ""
             logger.reason(f"LLM response finish_reason={finish_reason} content_len={len(content)} msg_keys={list(msg.keys())}")
             if not content:
                 logger.explore("LLM returned empty content", extra={"src": "executor", "finish_reason": finish_reason, "msg_keys": list(msg.keys()), "response_preview": str(data)[:2000]})

backend/src/plugins/translate/orchestrator.py

@@ -23,6 +23,7 @@ from collections.abc import Callable
 from datetime import UTC, datetime
 from typing import Any
 
+from sqlalchemy import text
 from sqlalchemy.orm import Session, joinedload, selectinload
 
 from ...core.config_manager import ConfigManager
@@ -72,6 +73,7 @@ class TranslationOrchestrator:
         job_id: str,
         is_scheduled: bool = False,
         trigger_type: str | None = None,
+        full_translation: bool = False,
     ) -> TranslationRun:
         with belief_scope("TranslationOrchestrator.start_run"):
             # Load and validate job
@@ -110,6 +112,7 @@ class TranslationOrchestrator:
                 "batch_size": job.batch_size,
                 "upsert_strategy": job.upsert_strategy,
                 "dictionary_ids": self._compute_dict_snapshot_hash(job_id),
+                "full_translation": full_translation,
             }
 
             # Compute key_hash from source_key_cols
@@ -249,6 +252,15 @@ class TranslationOrchestrator:
                 self.db.commit()
                 return run
 
+            # Check if executor cancelled itself due to cancellation flag
+            if run.status == "CANCELLED":
+                self._update_language_stats(run.id, language_stats_map)
+                self.db.commit()
+                logger.reflect("Run cancelled via cancellation flag", {
+                    "run_id": run.id,
+                })
+                return run
+
             # Aggregate per-language statistics after executor completes
             self._update_language_stats(run.id, language_stats_map)
 
@@ -746,7 +758,32 @@ class TranslationOrchestrator:
     # @SIDE_EFFECT: DB write; records event.
     def cancel_run(self, run_id: str) -> TranslationRun:
         with belief_scope("TranslationOrchestrator.cancel_run"):
-            run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
+            # Set short lock timeout to avoid blocking on row lock held by
+            # background executor thread (which holds RowExclusiveLock during
+            # batch processing). If the row is locked, we fall back to setting
+            # a cancellation flag via direct SQL UPDATE, which the executor
+            # checks after each batch commit.
+            try:
+                self.db.execute(text("SET LOCAL lock_timeout = '3s'"))
+                run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
+            except Exception:
+                # Row is locked — set cancellation flag via direct SQL
+                logger.explore("Row lock timeout — setting cancellation flag via direct SQL", {
+                    "run_id": run_id,
+                })
+                self.db.execute(
+                    text("UPDATE translation_runs SET error_message = 'CANCEL_REQUESTED' WHERE id = :run_id"),
+                    {"run_id": run_id},
+                )
+                self.db.commit()
+                run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
+                if not run:
+                    raise ValueError(f"Run '{run_id}' not found")
+                logger.reflect("Cancellation flag set for locked run", {
+                    "run_id": run_id,
+                })
+                return run
+
             if not run:
                 raise ValueError(f"Run '{run_id}' not found")
 

backend/src/plugins/translate/preview.py

@@ -17,7 +17,9 @@
 
 import hashlib
 import json
+import time
 import uuid
+from collections import defaultdict
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
@@ -169,10 +171,12 @@ class TranslationPreview:
         env_id: str | None = None,
     ) -> dict[str, Any]:
         with belief_scope("TranslationPreview.preview_rows"):
+            t0 = time.monotonic()
             logger.reason("Starting preview for job", {"job_id": job_id, "sample_size": sample_size})
 
             # 1. Load job
             job = self.db.query(TranslationJob).filter(TranslationJob.id == job_id).first()
+            logger.reason(f"TIMING: Load job: {time.monotonic() - t0:.2f}s", {"job_id": job_id})
             if not job:
                 raise ValueError(f"Translation job '{job_id}' not found")
             if not job.source_datasource_id:
@@ -201,6 +205,7 @@ class TranslationPreview:
                 sample_size=sample_size,
                 env_id=env_id,
             )
+            logger.reason(f"TIMING: Fetch sample rows: {time.monotonic() - t0:.2f}s", {"row_count": len(source_rows) if source_rows else 0})
             if not source_rows:
                 raise ValueError("No rows returned from datasource for preview")
 
@@ -341,11 +346,13 @@ class TranslationPreview:
                 "estimated_tokens": sample_total_tokens,
                 "max_tokens": max_tokens_for_call,
             })
+            t_llm = time.monotonic()
             llm_response = self._call_llm(
                 job=job,
                 prompt=prompt,
                 max_tokens=max_tokens_for_call,
             )
+            logger.reason(f"TIMING: LLM call: {time.monotonic() - t_llm:.2f}s response_len={len(llm_response) if llm_response else 0}", {})
 
             # 9. Parse LLM response (multi-language)
             translations = self._parse_llm_response(
@@ -508,6 +515,7 @@ class TranslationPreview:
                 "num_languages": num_languages,
                 "sample_cost": sample_cost,
             })
+            logger.reason(f"TIMING: Total preview: {time.monotonic() - t0:.2f}s", {"session_id": session.id, "row_count": actual_row_count})
             return result
     # endregion preview_rows
 
@@ -952,22 +960,31 @@ class TranslationPreview:
             if not api_key:
                 raise ValueError(f"Could not decrypt API key for provider '{job.provider_id}'")
 
-            # Build the API call based on provider type
             model = provider.default_model or "gpt-4o-mini"
             provider_type = provider.provider_type.lower() if provider.provider_type else "openai"
-
             disable_reasoning = getattr(job, 'disable_reasoning', False)
 
             if provider_type in ("openai", "openai_compatible", "openrouter", "kilo"):
-                response_text = self._call_openai_compatible(
-                    base_url=provider.base_url,
-                    api_key=api_key,
-                    model=model,
-                    prompt=prompt,
-                    provider_type=provider_type,
-                    max_tokens=max_tokens,
-                    disable_reasoning=disable_reasoning,
-                )
+                max_attempts = 2
+                last_error = None
+                for attempt in range(max_attempts):
+                    try:
+                        response_text = self._call_openai_compatible(
+                            base_url=provider.base_url,
+                            api_key=api_key,
+                            model=model,
+                            prompt=prompt,
+                            provider_type=provider_type,
+                            max_tokens=max_tokens * (attempt + 1),
+                            disable_reasoning=disable_reasoning or (attempt > 0),
+                        )
+                        break
+                    except ValueError as e:
+                        last_error = e
+                        if "empty content" in str(e) and attempt < max_attempts - 1:
+                            logger.explore("Empty LLM response, retrying with doubled max_tokens + force disable_reasoning", extra={"src": "preview", "attempt": attempt, "max_tokens": max_tokens * (attempt + 2)})
+                            continue
+                        raise
             else:
                 raise ValueError(f"Unsupported provider type '{provider_type}' for preview")
 
@@ -1011,17 +1028,18 @@ class TranslationPreview:
                 "temperature": 0.1,
                 "max_tokens": max_tokens,
             }
-            # Structured output (response_format) only for native OpenAI — upstream providers routed via
-            # Kilo/OpenRouter may not support it (e.g. StepFun returns "structured_outputs is not supported")
-            if provider_type in ("openai", "openai_compatible"):
+            # Structured output — Kilo gateway supports response_format, but upstream providers
+            # (e.g. StepFun) may reject it. We try with response_format and fall back on 400.
+            if provider_type in ("openai", "openai_compatible", "kilo", "openrouter"):
                 payload["response_format"] = {"type": "json_object"}
 
             # Suppress Chain of Thought reasoning to save output tokens
-            # Works universally via system prompt + API parameter for models that support it
+            # NOTE: Kilo/OpenRouter do NOT support disabling reasoning (returns 400)
             if disable_reasoning:
-                # Try multiple methods to suppress reasoning (varies by provider/deployment)
-                payload["reasoning_effort"] = "none"  # DeepSeek, Qwen
-                payload["extra_body"] = {"reasoning_effort": "none"}  # Kilo/OpenRouter proxy
+                # Kilo/OpenRouter reject reasoning_effort — only use for native OpenAI-compatible
+                if provider_type not in ("kilo", "openrouter"):
+                    payload["reasoning_effort"] = "none"
+                    payload["extra_body"] = {"reasoning_effort": "none"}
                 payload.pop("response_format", None)  # JSON mode triggers reasoning on some models
                 # Max tokens must be large enough for output even with some reasoning
                 payload["max_tokens"] = 8192
@@ -1036,7 +1054,13 @@ class TranslationPreview:
                 f"reasoning={'no' if disable_reasoning else 'yes'} "
                 f"prompt_len={len(prompt)}"
             )
-            response = http_requests.post(url, headers=headers, json=payload, timeout=120)
+
+            # ── Try request; retry once without response_format if upstream rejects it ──
+            response = http_requests.post(url, headers=headers, json=payload, timeout=600)
+            if not response.ok and response.status_code == 400 and "structured_outputs is not supported" in (response.text or ""):
+                logger.explore("Structured outputs not supported by upstream, retrying without response_format", extra={"src": "preview"})
+                payload.pop("response_format", None)
+                response = http_requests.post(url, headers=headers, json=payload, timeout=600)
             if not response.ok:
                 logger.explore(
                     f"LLM API error status={response.status_code} "
@@ -1051,10 +1075,34 @@ class TranslationPreview:
                 logger.explore("LLM returned no choices", extra={"src": "preview", "response_keys": list(data.keys()), "response_preview": str(data)[:2000]})
                 raise ValueError("LLM returned no choices")
 
-            finish_reason = choices[0].get("finish_reason", "none")
-            msg = choices[0].get("message", {})
-            content = msg.get("content", "")
+            try:
+                finish_reason = choices[0].get("finish_reason") or "none"
+                msg = choices[0].get("message") or {}
+                # Handle model refusal (content is empty/null, refusal field has reason)
+                refusal = msg.get("refusal")
+                if refusal:
+                    logger.explore("LLM refused to respond", extra={
+                        "src": "preview",
+                        "refusal": str(refusal)[:500],
+                        "finish_reason": finish_reason,
+                    })
+                    raise ValueError(f"LLM refused to respond: {refusal}")
+                content = msg.get("content") or ""
+            except TypeError as e:
+                logger.explore("TypeError processing LLM response", extra={
+                    "src": "preview_diag",
+                    "error": str(e),
+                    "choices_0_type": type(choices[0]).__name__,
+                    "choices_0_repr": repr(choices[0])[:2000],
+                    "finish_reason_raw": choices[0].get("finish_reason") if isinstance(choices[0], dict) else "N/A",
+                    "message_raw": choices[0].get("message") if isinstance(choices[0], dict) else "N/A",
+                    "data_type": type(data).__name__,
+                    "data_preview": str(data)[:2000],
+                })
+                raise ValueError(f"LLM response processing failed: {e}")
+
             logger.reason(f"LLM response finish_reason={finish_reason} content_len={len(content)} msg_keys={list(msg.keys())}")
+
             if not content:
                 logger.explore("LLM returned empty content", extra={"src": "preview", "finish_reason": finish_reason, "msg_keys": list(msg.keys()), "response_preview": str(data)[:2000]})
                 raise ValueError("LLM returned empty content")