diff --git a/backend/src/plugins/translate/__tests__/test_executor.py b/backend/src/plugins/translate/__tests__/test_executor.py
index 05a7b220..8dfaab90 100644
--- a/backend/src/plugins/translate/__tests__/test_executor.py
+++ b/backend/src/plugins/translate/__tests__/test_executor.py
@@ -15,7 +15,7 @@ from src.models.translate import (
     TranslationJob,
     TranslationRun,
 )
-from src.plugins.translate.executor import TranslationExecutor
+from src.plugins.translate.executor import TranslationExecutor, estimate_row_tokens
 
 
 # region mock_job [TYPE Function]
@@ -263,4 +263,435 @@ class TestCancellationFlag:
 
 
 # endregion TestCancellationFlag
+
+
+# region TestEstimateRowTokens [TYPE Class]
+# @PURPOSE: Tests for estimate_row_tokens — per-row token estimation for adaptive batch sizing.
+# @RELATION: BINDS_TO -> [estimate_row_tokens]
+class TestEstimateRowTokens:
+    """Unit tests for estimate_row_tokens()."""
+
+    # region test_empty_text [TYPE Function]
+    def test_empty_text(self) -> None:
+        """Empty source_text returns minimal tokens."""
+        job = MagicMock(spec=TranslationJob)
+        job.context_columns = []
+        tokens = estimate_row_tokens("", None, job)
+        assert tokens >= 1, f"Expected >= 1 token for empty text, got {tokens}"
+
+    # endregion test_empty_text
+
+    # region test_short_text [TYPE Function]
+    def test_short_text(self) -> None:
+        """Short ASCII text estimates roughly chars/2.2 tokens."""
+        job = MagicMock(spec=TranslationJob)
+        job.context_columns = []
+        tokens = estimate_row_tokens("Hello world", None, job)
+        # "Hello world" = 11 chars → ~5 tokens at 2.2 chars/token
+        assert 3 <= tokens <= 10, f"Expected reasonable token count, got {tokens}"
+
+    # endregion test_short_text
+
+    # region test_with_context [TYPE Function]
+    def test_with_context(self) -> None:
+        """Context columns contribute to token estimate."""
+        job = MagicMock(spec=TranslationJob)
+        job.context_columns = ["category", "description"]
+        source_data = {"category": "Billing", "description": "Monthly invoice summary"}
+        tokens_no_ctx = estimate_row_tokens("Product name", None, job)
+        tokens_with_ctx = estimate_row_tokens("Product name", source_data, job)
+        assert tokens_with_ctx > tokens_no_ctx, (
+            f"Context should increase tokens: no_ctx={tokens_no_ctx}, with_ctx={tokens_with_ctx}"
+        )
+
+    # endregion test_with_context
+
+    # region test_cjk_text [TYPE Function]
+    def test_cjk_text(self) -> None:
+        """CJK characters are token-denser (~1.5 chars/token)."""
+        job = MagicMock(spec=TranslationJob)
+        job.context_columns = []
+        # 12 CJK chars → 12/1.5 = 8 tokens, plus 1 for empty context = 9
+        tokens = estimate_row_tokens("你好世界这是一个测试消息", None, job)
+        assert tokens == 9, f"Expected 9 tokens (8 CJK + 1 empty ctx) for CJK, got {tokens}"
+
+    # endregion test_cjk_text
+
+    # region test_long_text [TYPE Function]
+    def test_long_text(self) -> None:
+        """Long text produces proportionally more tokens."""
+        job = MagicMock(spec=TranslationJob)
+        job.context_columns = []
+        long_text = "word " * 500  # ~2500 chars
+        tokens = estimate_row_tokens(long_text, None, job)
+        assert tokens > 100, f"Expected >100 tokens for long text, got {tokens}"
+
+    # endregion test_long_text
+
+    # region test_source_data_none_with_context_keys [TYPE Function]
+    def test_source_data_none_with_context_keys(self) -> None:
+        """When source_data is None but context_keys exist, no crash."""
+        job = MagicMock(spec=TranslationJob)
+        job.context_columns = ["category"]
+        tokens = estimate_row_tokens("Hello", None, job)
+        assert tokens >= 1, f"Expected at least 1 token, got {tokens}"
+
+    # endregion test_source_data_none_with_context_keys
+
+
+# endregion TestEstimateRowTokens
+
+
+# region TestAutoSizeBatches [TYPE Class]
+# @PURPOSE: Tests for _auto_size_batches — variable-sized batch splitting based on content length.
+# @RELATION: BINDS_TO -> [TranslationExecutor._auto_size_batches]
+# @RELATION: BINDS_TO -> [estimate_token_budget]
+class TestAutoSizeBatches:
+    """Tests for TranslationExecutor._auto_size_batches()."""
+
+    # region _make_executor [TYPE Function]
+    @pytest.fixture
+    def executor(self) -> TranslationExecutor:
+        db = MagicMock()
+        config_manager = MagicMock()
+        return TranslationExecutor(db, config_manager)
+
+    # endregion _make_executor
+
+    # region _make_job [TYPE Function]
+    @pytest.fixture
+    def job(self) -> MagicMock:
+        j = MagicMock(spec=TranslationJob)
+        j.id = "job-autosize-1"
+        j.translation_column = "source_text"
+        j.context_columns = []
+        j.target_languages = ["en"]
+        j.target_dialect = None
+        j.batch_size = 50
+        return j
+
+    # endregion _make_job
+
+    # region test_empty_rows [TYPE Function]
+    def test_empty_rows(self, executor: TranslationExecutor, job: MagicMock) -> None:
+        """Empty source_rows returns empty list."""
+        batches = executor._auto_size_batches(job, [], ["en"], provider_info="gpt-4o-mini")
+        assert batches == [], f"Expected empty list, got {batches}"
+
+    # endregion test_empty_rows
+
+    # region test_small_dataset_single_batch [TYPE Function]
+    @patch("src.plugins.translate.executor.estimate_token_budget")
+    def test_small_dataset_single_batch(
+        self,
+        mock_estimate: MagicMock,
+        executor: TranslationExecutor,
+        job: MagicMock,
+    ) -> None:
+        """Few short rows → single batch (all fit within budget)."""
+        mock_estimate.return_value = {
+            "batch_size_adjusted": 10,
+            "estimated_input_tokens": 5000,
+            "estimated_output_tokens": 2000,
+            "max_output_needed": 4096,
+            "warning": None,
+        }
+        source_rows = [
+            {"row_index": str(i), "source_text": "short text", "source_data": None}
+            for i in range(5)
+        ]
+        batches = executor._auto_size_batches(
+            job, source_rows, ["en"], provider_info="gpt-4o-mini",
+        )
+        assert len(batches) == 1, f"Expected 1 batch, got {len(batches)}"
+        assert len(batches[0]) == 5, f"Expected 5 rows in batch, got {len(batches[0])}"
+
+    # endregion test_small_dataset_single_batch
+
+    # region test_homogeneous_rows [TYPE Function]
+    @patch("src.plugins.translate.executor.estimate_token_budget")
+    def test_homogeneous_rows(
+        self,
+        mock_estimate: MagicMock,
+        executor: TranslationExecutor,
+        job: MagicMock,
+    ) -> None:
+        """Homogeneous short rows → many rows per batch (budget-efficient)."""
+        mock_estimate.return_value = {
+            "batch_size_adjusted": 20,
+            "estimated_input_tokens": 8000,
+            "estimated_output_tokens": 4000,
+            "max_output_needed": 4096,
+            "warning": None,
+        }
+        # 50 very short rows (10 chars each) → fits in ~3 batches of ~20
+        source_rows = [
+            {"row_index": str(i), "source_text": "short", "source_data": None}
+            for i in range(50)
+        ]
+        batches = executor._auto_size_batches(
+            job, source_rows, ["en"], provider_info="gpt-4o-mini",
+        )
+        # Should produce fewer batches than the fixed 50-size approach (1 batch)
+        # With budget ~8K tokens and rows ~2 tokens each → ~20-25 rows per batch
+        assert len(batches) >= 1, "Expected at least 1 batch"
+        total = sum(len(b) for b in batches)
+        assert total == 50, f"Expected 50 total rows, got {total}"
+        # Average batch size should be higher than 1
+        avg_size = sum(len(b) for b in batches) / len(batches)
+        assert avg_size >= 5, f"Expected avg batch size >= 5, got {avg_size}"
+
+    # endregion test_homogeneous_rows
+
+    # region test_mixed_length_rows [TYPE Function]
+    @patch("src.plugins.translate.executor.estimate_token_budget")
+    def test_mixed_length_rows(
+        self,
+        mock_estimate: MagicMock,
+        executor: TranslationExecutor,
+        job: MagicMock,
+    ) -> None:
+        """Mixed-length rows → variable-sized batches based on content length."""
+        # Tight budget: per_batch_budget = 1200 - 600 = 600 tokens
+        # Short rows (~1 token) can fit ~600 per batch.
+        # Long row (~1136 tokens) ALONE exceeds the per-batch budget → own batch.
+        mock_estimate.return_value = {
+            "batch_size_adjusted": 3,
+            "estimated_input_tokens": 1200,
+            "estimated_output_tokens": 500,
+            "max_output_needed": 4096,
+            "warning": None,
+        }
+        # Rows with very different lengths: 2 short, 1 long, 2 short
+        source_rows = [
+            {"row_index": "0", "source_text": "a", "source_data": None},
+            {"row_index": "1", "source_text": "b", "source_data": None},
+            # Long row with ~2500 chars (~1136 tokens)
+            {"row_index": "2", "source_text": "long " * 500, "source_data": None},
+            {"row_index": "3", "source_text": "c", "source_data": None},
+            {"row_index": "4", "source_text": "d", "source_data": None},
+        ]
+        batches = executor._auto_size_batches(
+            job, source_rows, ["en"], provider_info="gpt-4o-mini",
+        )
+        total = sum(len(b) for b in batches)
+        assert total == 5, f"Expected 5 total rows, got {total}"
+        # The long row exceeds the per-batch budget → isolated in own batch
+        # Short rows fit together → remaining 4 rows in 1-2 batches
+        assert len(batches) >= 2, (
+            f"Expected at least 2 batches (long row isolated), got {len(batches)}: "
+            f"{[len(b) for b in batches]}"
+        )
+        # Verify the long row is in its own batch
+        long_batches = [b for b in batches if any(r["source_text"] == "long " * 500 for r in b)]
+        assert len(long_batches) == 1, "Long row should be in exactly one batch"
+        assert len(long_batches[0]) == 1, "Long row batch should have exactly 1 row"
+
+    # endregion test_mixed_length_rows
+
+    # region test_row_exceeds_budget [TYPE Function]
+    @patch("src.plugins.translate.executor.estimate_token_budget")
+    def test_row_exceeds_budget(
+        self,
+        mock_estimate: MagicMock,
+        executor: TranslationExecutor,
+        job: MagicMock,
+    ) -> None:
+        """Single row exceeding per-batch budget → placed in own batch with WARN."""
+        mock_estimate.return_value = {
+            "batch_size_adjusted": 5,
+            "estimated_input_tokens": 3000,
+            "estimated_output_tokens": 2000,
+            "max_output_needed": 4096,
+            "warning": None,
+        }
+        # Row with 3000 chars text — token count will exceed the per_batch_budget
+        # which is estimated_input_tokens - PROMPT_BASE_TOKENS = 3000 - 600 = 2400
+        # The row text "x" * 3000 has ~3000/2.2 ≈ 1364 tokens → exceeds 2400...
+        # Actually 1364 < 2400, so it would fit. Let me use longer text.
+        # "x" * 6000 → ~6000/2.2 ≈ 2727 tokens > 2400
+        source_rows = [
+            {"row_index": "0", "source_text": "x" * 6000, "source_data": None},
+            {"row_index": "1", "source_text": "short", "source_data": None},
+            {"row_index": "2", "source_text": "tiny", "source_data": None},
+        ]
+        batches = executor._auto_size_batches(
+            job, source_rows, ["en"], provider_info="gpt-4o-mini",
+        )
+        total = sum(len(b) for b in batches)
+        assert total == 3, f"Expected 3 total rows, got {total}"
+        # First row should be in its own batch
+        assert len(batches[0]) == 1, "Expected oversized row in own batch"
+
+    # endregion test_row_exceeds_budget
+
+    # region test_budget_failure_fallback [TYPE Function]
+    @patch("src.plugins.translate.executor.estimate_token_budget")
+    def test_budget_failure_fallback(
+        self,
+        mock_estimate: MagicMock,
+        executor: TranslationExecutor,
+        job: MagicMock,
+    ) -> None:
+        """When estimate_token_budget fails (recommended=0), fallback to fixed batch_size."""
+        mock_estimate.return_value = {
+            "batch_size_adjusted": 0,  # Failure signal
+            "estimated_input_tokens": 0,
+            "estimated_output_tokens": 0,
+            "max_output_needed": 0,
+            "warning": "Budget error",
+        }
+        source_rows = [
+            {"row_index": str(i), "source_text": "short text", "source_data": None}
+            for i in range(60)
+        ]
+        batches = executor._auto_size_batches(
+            job, source_rows, ["en"], provider_info="gpt-4o-mini",
+        )
+        # Fallback to job.batch_size=50 → should produce 2 batches (50 + 10)
+        assert len(batches) == 2, f"Expected 2 fallback batches (50+10), got {len(batches)}"
+        assert len(batches[0]) == 50, f"Expected 50 in first fallback batch, got {len(batches[0])}"
+
+    # endregion test_budget_failure_fallback
+
+    # region test_budget_zero_input_collapse [TYPE Function]
+    @patch("src.plugins.translate.executor.estimate_token_budget")
+    def test_budget_zero_input_collapse(
+        self,
+        mock_estimate: MagicMock,
+        executor: TranslationExecutor,
+        job: MagicMock,
+    ) -> None:
+        """When per_batch_budget collapses (<=0), fallback to fixed batch_size."""
+        mock_estimate.return_value = {
+            "batch_size_adjusted": 5,
+            "estimated_input_tokens": 100,  # Less than PROMPT_BASE_TOKENS (600)
+            "estimated_output_tokens": 50,
+            "max_output_needed": 100,
+            "warning": None,
+        }
+        source_rows = [
+            {"row_index": str(i), "source_text": "test", "source_data": None}
+            for i in range(60)
+        ]
+        batches = executor._auto_size_batches(
+            job, source_rows, ["en"], provider_info="gpt-4o-mini",
+        )
+        # Should fallback to job.batch_size=50
+        assert len(batches) == 2, f"Expected 2 fallback batches, got {len(batches)}"
+
+    # endregion test_budget_zero_input_collapse
+
+    # region test_provider_info_resolution [TYPE Function]
+    @patch("src.plugins.translate.executor.estimate_token_budget")
+    def test_provider_info_resolution(
+        self,
+        mock_estimate: MagicMock,
+        executor: TranslationExecutor,
+        job: MagicMock,
+    ) -> None:
+        """When provider_info is None, _resolve_provider_model is called."""
+        mock_estimate.return_value = {
+            "batch_size_adjusted": 10,
+            "estimated_input_tokens": 5000,
+            "estimated_output_tokens": 2000,
+            "max_output_needed": 4096,
+            "warning": None,
+        }
+        source_rows = [
+            {"row_index": "0", "source_text": "hello", "source_data": None},
+        ]
+        # provider_info=None → should call _resolve_provider_model
+        with patch.object(executor, '_resolve_provider_model', return_value="gpt-4o-mini") as mock_resolve:
+            batches = executor._auto_size_batches(
+                job, source_rows, ["en"], provider_info=None,
+            )
+            mock_resolve.assert_called_once_with(job)
+            assert len(batches) == 1
+
+    # endregion test_provider_info_resolution
+
+    # region test_resolve_provider_model [TYPE Function]
+    def test_resolve_provider_model_no_provider(self, executor: TranslationExecutor) -> None:
+        """_resolve_provider_model returns None when job has no provider_id."""
+        job = MagicMock(spec=TranslationJob)
+        job.provider_id = None
+        result = executor._resolve_provider_model(job)
+        assert result is None, f"Expected None when no provider_id, got {result}"
+
+    # endregion test_resolve_provider_model
+
+    # region test_resolve_provider_model_with_provider [TYPE Function]
+    @patch("src.plugins.translate.executor.LLMProviderService")
+    def test_resolve_provider_model_with_provider(
+        self,
+        mock_provider_svc: MagicMock,
+        executor: TranslationExecutor,
+    ) -> None:
+        """_resolve_provider_model returns the default model name."""
+        job = MagicMock(spec=TranslationJob)
+        job.provider_id = "provider-1"
+        mock_svc_instance = MagicMock()
+        mock_provider = MagicMock()
+        mock_provider.default_model = "gpt-4o-mini"
+        mock_svc_instance.get_provider.return_value = mock_provider
+        mock_provider_svc.return_value = mock_svc_instance
+
+        executor.db = MagicMock()
+        result = executor._resolve_provider_model(job)
+        assert result == "gpt-4o-mini", f"Expected 'gpt-4o-mini', got {result}"
+
+    # endregion test_resolve_provider_model_with_provider
+
+    # region test_resolve_provider_model_exception [TYPE Function]
+    @patch("src.plugins.translate.executor.LLMProviderService")
+    def test_resolve_provider_model_exception(
+        self,
+        mock_provider_svc: MagicMock,
+        executor: TranslationExecutor,
+    ) -> None:
+        """_resolve_provider_model returns None on exception."""
+        job = MagicMock(spec=TranslationJob)
+        job.provider_id = "provider-1"
+        mock_svc_instance = MagicMock()
+        mock_svc_instance.get_provider.side_effect = Exception("DB error")
+        mock_provider_svc.return_value = mock_svc_instance
+
+        executor.db = MagicMock()
+        result = executor._resolve_provider_model(job)
+        assert result is None, f"Expected None on exception, got {result}"
+
+    # endregion test_resolve_provider_model_exception
+
+    # region test_at_least_one_row_per_batch [TYPE Function]
+    @patch("src.plugins.translate.executor.estimate_token_budget")
+    def test_at_least_one_row_per_batch(
+        self,
+        mock_estimate: MagicMock,
+        executor: TranslationExecutor,
+        job: MagicMock,
+    ) -> None:
+        """Even when rows are large, each batch has at least 1 row."""
+        mock_estimate.return_value = {
+            "batch_size_adjusted": 1,
+            "estimated_input_tokens": 1000,
+            "estimated_output_tokens": 500,
+            "max_output_needed": 4096,
+            "warning": None,
+        }
+        source_rows = [
+            {"row_index": str(i), "source_text": "large " * 200, "source_data": None}
+            for i in range(3)
+        ]
+        batches = executor._auto_size_batches(
+            job, source_rows, ["en"], provider_info="gpt-4o-mini",
+        )
+        total = sum(len(b) for b in batches)
+        assert total == 3, f"Expected 3 total rows, got {total}"
+        assert all(len(b) >= 1 for b in batches), "Each batch must have at least 1 row"
+
+    # endregion test_at_least_one_row_per_batch
+
+
+# endregion TestAutoSizeBatches
 # endregion ExecutorTests
diff --git a/backend/src/plugins/translate/_token_budget.py b/backend/src/plugins/translate/_token_budget.py
index ccd12fd0..d9f1131b 100644
--- a/backend/src/plugins/translate/_token_budget.py
+++ b/backend/src/plugins/translate/_token_budget.py
@@ -14,8 +14,8 @@ DEFAULT_CONTEXT_WINDOW = 64000
 # #endregion DEFAULT_CONTEXT_WINDOW
 
 # #region DEFAULT_MAX_OUTPUT_TOKENS [TYPE Constant]
-# @BRIEF Default max_tokens setting for LLM output (8192 tokens).
-DEFAULT_MAX_OUTPUT_TOKENS = 8192
+# @BRIEF Default max_tokens setting for LLM output (16384 — sufficient for 50 rows x 4 languages).
+DEFAULT_MAX_OUTPUT_TOKENS = 16384
 # #endregion DEFAULT_MAX_OUTPUT_TOKENS
 
 # #region REASONING_OVERHEAD [TYPE Constant]
@@ -23,14 +23,37 @@ DEFAULT_MAX_OUTPUT_TOKENS = 8192
 REASONING_OVERHEAD = 2000
 # #endregion REASONING_OVERHEAD
 
+# #region PROVIDER_DEFAULTS [TYPE Constant]
+# @BRIEF Provider-aware defaults for context_window and max_output_tokens.
+# Maps model name (or "default" fallback) to capacity limits.
+# @RATIONALE: Different providers have drastically different context windows and
+#             output limits. Using a single default for all causes either wasted
+#             capacity (underestimation) or truncation (overestimation).
+PROVIDER_DEFAULTS: dict[str, dict[str, int]] = {
+    "gpt-4o-mini": {"context_window": 128000, "max_output_tokens": 16384},
+    "gpt-4o": {"context_window": 128000, "max_output_tokens": 16384},
+    "o1-mini": {"context_window": 128000, "max_output_tokens": 65536},
+    "claude-3-5-sonnet": {"context_window": 200000, "max_output_tokens": 8192},
+    "deepseek-v4-flash": {"context_window": 64000, "max_output_tokens": 8192},
+    "default": {"context_window": 64000, "max_output_tokens": 16384},
+}
+# #endregion PROVIDER_DEFAULTS
+
 # #region OUTPUT_PER_ROW_PER_LANG [TYPE Constant]
 # @BRIEF Estimated output tokens per row per language in JSON response format.
-OUTPUT_PER_ROW_PER_LANG = 60
+#            Increased from 60 to 120 because SQL/dashboard text and JSON structure need more.
+OUTPUT_PER_ROW_PER_LANG = 120
 # #endregion OUTPUT_PER_ROW_PER_LANG
 
+# #region JSON_OVERHEAD_PER_ROW [TYPE Constant]
+# @BRIEF Estimated overhead tokens for JSON keys, brackets, and formatting per row.
+JSON_OVERHEAD_PER_ROW = 50
+# #endregion JSON_OVERHEAD_PER_ROW
+
 # #region PROMPT_BASE_TOKENS [TYPE Constant]
-# @BRIEF Base tokens for system prompt + instructions + JSON format specification.
-PROMPT_BASE_TOKENS = 300
+# @BRIEF Base tokens for system prompt + instructions + dictionary section + JSON format specification.
+#            Increased from 300 to 600 to account for longer template, system msg, and dict preamble.
+PROMPT_BASE_TOKENS = 600
 # #endregion PROMPT_BASE_TOKENS
 
 # #region DICT_TOKENS_PER_ENTRY [TYPE Constant]
@@ -54,15 +77,47 @@ MIN_MAX_TOKENS = 4096
 # #endregion MIN_MAX_TOKENS
 
 # #region MAX_OUTPUT_HEADROOM [TYPE Constant]
-# @BRIEF Extra headroom added to max_output_needed beyond the estimate (1000 tokens buffer).
-MAX_OUTPUT_HEADROOM = 1000
+# @BRIEF Extra headroom added to max_output_needed beyond the estimate (3000 = 10-20% for variance).
+#            Increased from 1000 to 3000 because SQL/dashboard text output varies significantly.
+MAX_OUTPUT_HEADROOM = 3000
 # #endregion MAX_OUTPUT_HEADROOM
 
 
+# region _estimate_tokens_for_text [TYPE Function]
+# @BRIEF Estimate token count for a text string with CJK-aware heuristics.
+#        CJK characters (~1.5 chars/token) vs other text (~2.2 chars/token).
+# @PRE: text is a string.
+# @POST: Returns estimated token count >= 1.
+# @RATIONALE: CJK characters are more token-dense than Latin/Cyrillic text.
+#             Using a single ratio undercounts CJK input and causes truncation.
+# @REJECTED: Using tiktoken — would introduce a heavy dependency for estimation only.
+def _estimate_tokens_for_text(text: str) -> int:
+    """Estimate token count with CJK-aware heuristics.
+
+    CJK characters (CJK Unified Ideographs) use ~1.5 chars/token.
+    All other characters use ~2.2 chars/token.
+    """
+    if not text:
+        return 1
+
+    cjk_count = 0
+    other_count = 0
+    for ch in text:
+        if '\u4e00' <= ch <= '\u9fff' or '\u3000' <= ch <= '\u303f' or '\uff00' <= ch <= '\uffef':
+            cjk_count += 1
+        else:
+            other_count += 1
+
+    cjk_tokens = cjk_count / 1.5 if cjk_count else 0
+    other_tokens = other_count / 2.2 if other_count else 0
+    return max(1, int(cjk_tokens + other_tokens))
+# endregion _estimate_tokens_for_text
+
+
 # region _count_rows_that_fit [TYPE Function]
 # @BRIEF Count how many rows fit within the available input budget.
 # @PRE: input_per_row is non-empty; available_budget > 0.
-# @POST: Returns (safe_count, total_input_tokens).
+# @POST: Returns (safe_count, total_input_tokens). When no rows fit, returns (0, 0).
 def _count_rows_that_fit(
     input_per_row: list[int],
     available_budget: int,
@@ -71,6 +126,7 @@ def _count_rows_that_fit(
 
     Returns:
         (safe_count, total_input_tokens): Number of rows and their total tokens.
+        Returns (0, 0) when the first row alone does not fit (MEDIUM fix).
     """
     running_total = 0
     safe_size = 0
@@ -80,7 +136,8 @@ def _count_rows_that_fit(
             safe_size += 1
         else:
             break
-    safe_size = max(safe_size, 1)
+    # MEDIUM: Return (0, 0) when no rows fit — signal upstream that oversizing occurred.
+    # Prevents silent clamping to 1 which would produce truncated LLM calls.
     return safe_size, running_total
 # endregion _count_rows_that_fit
 
@@ -93,6 +150,64 @@ def _count_rows_that_fit(
 #             depends on the LLM model. Estimates are intentionally conservative to prevent truncation.
 # @REJECTED: Using tiktoken or similar tokenizer — would introduce a heavy dependency and still
 #            not match DeepSeek's tokenizer exactly.
+def _calculate_output_tokens(
+    safe_size: int,
+    num_languages: int,
+) -> int:
+    """Calculate estimated output tokens for a batch."""
+    return (
+        safe_size * num_languages * OUTPUT_PER_ROW_PER_LANG
+        + safe_size * JSON_OVERHEAD_PER_ROW
+        + REASONING_OVERHEAD
+    )
+
+
+def _apply_output_aware_batch_sizing(
+    safe_size: int,
+    num_languages: int,
+    max_output_tokens: int,
+) -> int:
+    """Reduce batch size until estimated output fits within max_output_tokens."""
+    while safe_size > 0:
+        needed_output = (
+            safe_size * num_languages * OUTPUT_PER_ROW_PER_LANG
+            + safe_size * JSON_OVERHEAD_PER_ROW
+            + REASONING_OVERHEAD
+            + MAX_OUTPUT_HEADROOM
+        )
+        if needed_output <= max_output_tokens:
+            break
+        safe_size -= 1
+    return safe_size
+
+
+def _build_warning(
+    batch_size: int | None,
+    safe_size: int,
+    total_rows: int,
+    context_window: int,
+    estimated_input: int,
+    max_output_needed: int,
+    dict_warning: str | None,
+) -> str | None:
+    """Build warning message when batch size is reduced."""
+    warning = None
+    if batch_size and safe_size < batch_size:
+        total_estimated = estimated_input + max_output_needed
+        warning = (
+            f"Reduced batch size from {batch_size} to {safe_size} "
+            f"(estimated {total_estimated} tokens vs {context_window} window)"
+        )
+    elif not batch_size and safe_size < total_rows:
+        warning = (
+            f"Auto-calculated batch size of {safe_size} from {total_rows} rows "
+            f"(output-limited)"
+        )
+    if dict_warning:
+        warning = f"{warning}; {dict_warning}" if warning else dict_warning
+    return warning
+
+
 def estimate_token_budget(
     source_rows: list[dict],
     target_languages: list[str],
@@ -100,8 +215,9 @@ def estimate_token_budget(
     context_columns: list[str] | None = None,
     dictionary_entries: list | None = None,
     batch_size: int | None = None,
-    context_window: int = DEFAULT_CONTEXT_WINDOW,
-    max_output_tokens: int = DEFAULT_MAX_OUTPUT_TOKENS,
+    context_window: int | None = None,
+    max_output_tokens: int | None = None,
+    provider_info: str | None = None,
 ) -> dict:
     """Estimate token budget and return safe batch parameters.
 
@@ -112,8 +228,11 @@ def estimate_token_budget(
         context_columns: Optional list of keys for context columns.
         dictionary_entries: Optional list of dictionary entries for glossary.
         batch_size: Desired batch size. If None, auto-calculate max safe size.
-        context_window: Model context window (default 64000 for DeepSeek v4 Flash).
-        max_output_tokens: Hard max output tokens limit (default 8192).
+        context_window: Model context window. If None, resolved from provider_info.
+        max_output_tokens: Hard max output tokens limit. If None, resolved from provider_info.
+        provider_info: Optional provider model name (e.g. "gpt-4o-mini") for provider-aware defaults.
+                       When provided without explicit context_window/max_output_tokens, uses
+                       PROVIDER_DEFAULTS to set appropriate limits.
 
     Returns:
         dict with:
@@ -126,31 +245,47 @@ def estimate_token_budget(
     if not target_languages:
         target_languages = ["en"]
 
+    # Resolve provider-aware defaults
+    if provider_info and context_window is None and max_output_tokens is None:
+        provider_settings = PROVIDER_DEFAULTS.get(provider_info)
+        if provider_settings:
+            context_window = provider_settings["context_window"]
+            max_output_tokens = provider_settings["max_output_tokens"]
+
+    # Fall back to module defaults if still None
+    if context_window is None:
+        context_window = DEFAULT_CONTEXT_WINDOW
+    if max_output_tokens is None:
+        max_output_tokens = DEFAULT_MAX_OUTPUT_TOKENS
+
     num_languages = len(target_languages)
 
-    # 1. Estimate tokens per row
+    # 1. Estimate tokens per row using CJK-aware heuristics
     input_per_row = []
     limit = batch_size if batch_size else len(source_rows)
     for i, row in enumerate(source_rows):
         if i >= limit:
             break
         text = str(row.get(source_column, "") or "")
-        # Use ~2.2 chars per token for mixed Russian/English text
-        estimated_tokens = max(1, int(len(text) / CHARS_PER_TOKEN_MIXED))
-        # Add context columns
+        estimated_tokens = _estimate_tokens_for_text(text)
         if context_columns:
             for col in context_columns:
                 val = str(row.get(col, "") or "")
-                estimated_tokens += max(1, int(len(val) / CHARS_PER_TOKEN_MIXED))
+                estimated_tokens += _estimate_tokens_for_text(val)
         input_per_row.append(estimated_tokens)
 
-    # 2. Calculate dictionary tokens
+    # 2. Calculate dictionary tokens with warning if capped
     dict_tokens = 0
+    dict_warning = None
     if dictionary_entries:
-        dict_tokens = min(
-            len(dictionary_entries) * DICT_TOKENS_PER_ENTRY,
-            DICT_TOKENS_MAX,
-        )
+        raw_dict_tokens = len(dictionary_entries) * DICT_TOKENS_PER_ENTRY
+        dict_tokens = min(raw_dict_tokens, DICT_TOKENS_MAX)
+        if raw_dict_tokens > DICT_TOKENS_MAX:
+            dict_warning = (
+                f"Dictionary entries ({len(dictionary_entries)} entries "
+                f"≈ {raw_dict_tokens} tokens) exceed cap of {DICT_TOKENS_MAX} — "
+                f"truncated to {DICT_TOKENS_MAX} in estimation"
+            )
 
     prompt_tokens = PROMPT_BASE_TOKENS + dict_tokens
     available_input_budget = context_window - max_output_tokens
@@ -161,19 +296,26 @@ def estimate_token_budget(
 
     # If batch_size was specified and we reduced it, recalculate
     if batch_size and safe_size < batch_size:
-        # Ensure at minimum one row even for huge content
-        safe_size = max(safe_size, 1)
         _, truncated_total = _count_rows_that_fit(
             input_per_row[:batch_size], available_input_budget,
         )
         estimated_input = prompt_tokens + truncated_total
 
-    # 4. Estimate output tokens
-    estimated_output = (
-        safe_size * num_languages * OUTPUT_PER_ROW_PER_LANG + REASONING_OVERHEAD
+    # 4. Estimate output tokens (includes JSON overhead per row)
+    estimated_output = _calculate_output_tokens(safe_size, num_languages)
+
+    # 5. Output-aware batch sizing: reduce batch if output exceeds limit
+    safe_size = _apply_output_aware_batch_sizing(
+        safe_size, num_languages, max_output_tokens,
     )
 
-    # 5. Calculate recommended max_tokens
+    # Recalculate totals after output-aware reduction
+    if safe_size > 0:
+        new_input_total = sum(input_per_row[:safe_size])
+        estimated_input = prompt_tokens + new_input_total
+        estimated_output = _calculate_output_tokens(safe_size, num_languages)
+
+    # 6. Calculate recommended max_tokens
     max_output_needed = min(
         estimated_output + MAX_OUTPUT_HEADROOM,
         context_window - estimated_input,
@@ -181,14 +323,12 @@ def estimate_token_budget(
     max_output_needed = max(max_output_needed, MIN_MAX_TOKENS)
     max_output_needed = min(max_output_needed, max_output_tokens)
 
-    # 6. Generate warning if batch was reduced
-    warning = None
-    if batch_size and safe_size < batch_size:
-        total_estimated = estimated_input + max_output_needed
-        warning = (
-            f"Reduced batch size from {batch_size} to {safe_size} "
-            f"(estimated {total_estimated} tokens vs {context_window} window)"
-        )
+    # 7. Generate warning if batch was reduced
+    warning = _build_warning(
+        batch_size, safe_size, len(source_rows),
+        context_window, estimated_input, max_output_needed,
+        dict_warning,
+    )
 
     return {
         "batch_size_adjusted": safe_size,
diff --git a/backend/src/plugins/translate/executor.py b/backend/src/plugins/translate/executor.py
index 8a9b59fd..3796f69c 100644
--- a/backend/src/plugins/translate/executor.py
+++ b/backend/src/plugins/translate/executor.py
@@ -40,7 +40,7 @@ from ...models.translate import (
 )
 from ...services.llm_prompt_templates import render_prompt
 from ...services.llm_provider import LLMProviderService
-from ._token_budget import DEFAULT_CONTEXT_WINDOW, DEFAULT_MAX_OUTPUT_TOKENS, estimate_token_budget
+from ._token_budget import estimate_token_budget
 from .dictionary import DictionaryManager
 from .preview import DEFAULT_EXECUTION_PROMPT_TEMPLATE
 from .prompt_builder import ContextAwarePromptBuilder
@@ -194,6 +194,35 @@ def _check_translation_cache(
 # #endregion _check_translation_cache
 
 
+# #region estimate_row_tokens [C:2] [TYPE Function] [SEMANTICS translate, token, estimation]
+# @BRIEF Estimate token count for a single source row including context fields.
+# @PRE: source_text is a string.
+# @POST: Returns estimated token count >= 1.
+# @RELATION DEPENDS_ON -> [_estimate_tokens_for_text]
+def estimate_row_tokens(
+    source_text: str,
+    source_data: dict | None,
+    job,
+) -> int:
+    """Estimate token count for a single source row including context fields.
+
+    Uses CJK-aware heuristics via _token_budget._estimate_tokens_for_text.
+    Context fields from job.context_columns are included in the estimate.
+    """
+    from ._token_budget import _estimate_tokens_for_text
+
+    text_tokens = _estimate_tokens_for_text(source_text or "")
+
+    context_keys = job.context_columns or []
+    ctx_text = ""
+    if source_data and context_keys:
+        ctx_text = " ".join(str(source_data.get(k, "")) for k in context_keys)
+    ctx_tokens = _estimate_tokens_for_text(ctx_text)
+
+    return text_tokens + ctx_tokens
+# #endregion estimate_row_tokens
+
+
 # #region TranslationExecutor [C:4] [TYPE Class]
 # @BRIEF Process translation batches: fetch source rows, filter dict, call LLM, persist results.
 # @PRE: DB session and config manager available.
@@ -215,6 +244,179 @@ class TranslationExecutor:
         self._current_run_id: str | None = None
         self._preview_edits_cache: dict[str, dict[str, str]] | None = None  # key_hash -> {lang_code: edited_value}
 
+    # region _resolve_provider_model [TYPE Function]
+    # @BRIEF Resolve the LLM provider model name for token budget estimation.
+    # @POST: Returns model name string or None if resolution fails.
+    # @SIDE_EFFECT: DB query to LLM provider table.
+    def _resolve_provider_model(self, job) -> str | None:
+        """Resolve the provider model name for token budget estimation."""
+        if not job.provider_id:
+            return None
+        try:
+            p_svc = LLMProviderService(self.db)
+            p = p_svc.get_provider(job.provider_id)
+            if p:
+                return p.default_model or "gpt-4o-mini"
+        except Exception:
+            pass
+        return None
+    # endregion _resolve_provider_model
+
+    # region _auto_size_batches [TYPE Function]
+    # @PURPOSE: Split source rows into variable-sized batches based on actual content length.
+    #           Uses estimate_token_budget to determine safe per-batch budgets and
+    #           _count_rows_that_fit logic to greedily fill each batch.
+    # @PRE: source_rows is non-empty. job has valid config.
+    # @POST: Returns list of batches, each batch is a list of row dicts.
+    #         Each batch fits within the estimated token budget for its rows.
+    # @SIDE_EFFECT: DB query to resolve provider model. Logs batch statistics.
+    # @RATIONALE: Fixed batch_size of 50 wastes LLM context for short rows and
+    #             overflows for long rows. Variable sizing adapts to row content length,
+    #             maximizing throughput while preventing truncation.
+    # @REJECTED: Fixed batch_size of 50 — causes truncation on long-content rows.
+    #            Single monolithic batch — would lose all progress on any failure.
+    def _auto_size_batches(
+        self,
+        job,
+        source_rows: list[dict],
+        target_languages: list[str],
+        provider_info: str | None = None,
+    ) -> list[list[dict]]:
+        """Split source rows into auto-sized batches based on content length.
+
+        Each batch is sized so that its total estimated tokens fit within the
+        available context window (input budget), accounting for prompt overhead,
+        dictionary entries, and output tokens.
+
+        Returns:
+            List of batches, each batch is a list of row dicts.
+        """
+        with belief_scope("TranslationExecutor._auto_size_batches"):
+            if not source_rows:
+                return []
+
+            # Resolve provider info if not provided
+            if provider_info is None:
+                provider_info = self._resolve_provider_model(job)
+
+            # 1. Estimate per-row token counts
+            row_tokens: list[int] = []
+            for row in source_rows:
+                source_text = row.get("source_text", "")
+                source_data = row.get("source_data")
+                tokens = estimate_row_tokens(source_text, source_data, job)
+                row_tokens.append(tokens)
+
+            # 2. Get budget recommendation from estimate_token_budget
+            #    Pass all rows to get a global safe batch size estimate.
+            budget = estimate_token_budget(
+                source_rows=source_rows,
+                target_languages=target_languages,
+                source_column=job.translation_column or "source_text",
+                context_columns=job.context_columns,
+                batch_size=len(source_rows),
+                provider_info=provider_info,
+            )
+
+            recommended = budget.get("batch_size_adjusted", 0)
+
+            # Fallback: if budget calculation fails or returns zero, use fixed size
+            if recommended <= 0:
+                fallback_size = job.batch_size or 50
+                logger.explore("Token budget returned zero — falling back to fixed batch size", {
+                    "fallback_size": fallback_size,
+                    "total_rows": len(source_rows),
+                })
+                return [
+                    source_rows[i:i + fallback_size]
+                    for i in range(0, len(source_rows), fallback_size)
+                ]
+
+            # 3. Compute per-batch row-content budget
+            #    estimated_input_tokens includes PROMPT_BASE_TOKENS + dict_tokens +
+            #    sum of row tokens for the first `recommended` rows.
+            #    We subtract the prompt overhead to get the per-batch row budget.
+            from ._token_budget import PROMPT_BASE_TOKENS
+
+            estimated_input = budget.get("estimated_input_tokens", 50000)
+            per_batch_budget = estimated_input - PROMPT_BASE_TOKENS
+
+            # Safety: if budget computation collapsed, fall back
+            if per_batch_budget <= 0:
+                fallback_size = job.batch_size or 50
+                logger.explore("Per-batch budget collapsed — falling back to fixed batch size", {
+                    "estimated_input": estimated_input,
+                    "prompt_base": PROMPT_BASE_TOKENS,
+                    "fallback_size": fallback_size,
+                })
+                return [
+                    source_rows[i:i + fallback_size]
+                    for i in range(0, len(source_rows), fallback_size)
+                ]
+
+            # 4. Greedy batch splitting: accumulate rows until the next row would
+            #    exceed the per-batch token budget. Use output-aware hard cap too:
+            #    a single batch never exceeds 2x the recommended batch size as
+            #    a safety net against wildly oversized estimates.
+            max_rows_hard_cap = max(recommended * 2, 20)
+            batches: list[list[dict]] = []
+            current_batch: list[dict] = []
+            current_tokens = 0
+
+            for i, row in enumerate(source_rows):
+                rt = row_tokens[i]
+
+                # ── Edge case: single row exceeds entire per-batch budget ──
+                if rt > per_batch_budget:
+                    # Flush current batch first
+                    if current_batch:
+                        batches.append(current_batch)
+                        current_batch = []
+                        current_tokens = 0
+                    logger.reason("Single row exceeds per-batch token budget — placing in own batch", {
+                        "row_index": i,
+                        "row_tokens": rt,
+                        "per_batch_budget": per_batch_budget,
+                    })
+                    batches.append([row])
+                    continue
+
+                # ── Start a new batch if: ──
+                #   a) Batch is non-empty AND adding this row would exceed budget
+                #   b) Hard row-count cap reached
+                should_split = bool(
+                    current_batch
+                    and (current_tokens + rt > per_batch_budget
+                         or len(current_batch) >= max_rows_hard_cap)
+                )
+
+                if should_split:
+                    batches.append(current_batch)
+                    current_batch = [row]
+                    current_tokens = rt
+                else:
+                    current_batch.append(row)
+                    current_tokens += rt
+
+            if current_batch:
+                batches.append(current_batch)
+
+            # 5. Log adaptive batch statistics
+            if batches:
+                avg_size = sum(len(b) for b in batches) / max(1, len(batches))
+                max_size = max(len(b) for b in batches)
+                logger.reason("Auto-sized batches", {
+                    "total_rows": len(source_rows),
+                    "num_batches": len(batches),
+                    "avg_batch_size": round(avg_size, 1),
+                    "max_batch_size": max_size,
+                    "recommended_batch_size": recommended,
+                    "per_batch_budget": per_batch_budget,
+                })
+
+            return batches
+    # endregion _auto_size_batches
+
     # region execute_run [TYPE Function]
     # @PURPOSE: Run full translation execution for a TranslationRun.
     # @PRE: run is in PENDING or RUNNING status with valid job config.
@@ -234,7 +436,7 @@ class TranslationExecutor:
             logger.reason("Starting translation execution", {
                 "run_id": run.id,
                 "job_id": job.id,
-                "batch_size": job.batch_size,
+                "configured_batch_size": job.batch_size,
             })
 
             # Load preview edits for carry-forward
@@ -276,17 +478,26 @@ class TranslationExecutor:
             total_rows = len(source_rows)
             run.total_records = total_rows
 
-            # Split into batches
-            batch_size = job.batch_size or 50
-            batches = [
-                source_rows[i:i + batch_size]
-                for i in range(0, total_rows, batch_size)
-            ]
+            # Resolve target languages for batch sizing
+            target_languages = job.target_languages or [job.target_dialect or "en"]
+            if not isinstance(target_languages, list):
+                target_languages = [str(target_languages)]
+
+            # Split into auto-sized batches based on content length
+            provider_model = self._resolve_provider_model(job)
+            batches = self._auto_size_batches(
+                job=job,
+                source_rows=source_rows,
+                target_languages=target_languages,
+                provider_info=provider_model,
+            )
 
             logger.reason(f"Processing {len(batches)} batches", {
                 "run_id": run.id,
                 "total_rows": total_rows,
-                "batch_size": batch_size,
+                "num_batches": len(batches),
+                "avg_batch_size": round(sum(len(b) for b in batches) / max(1, len(batches)), 1),
+                "max_batch_size": max(len(b) for b in batches) if batches else 0,
             })
 
             successful_records = 0
@@ -841,6 +1052,17 @@ class TranslationExecutor:
 
             # Process rows needing LLM translation
             if rows_for_llm:
+                # Resolve provider model for provider-aware token budget
+                provider_model = None
+                if job.provider_id:
+                    try:
+                        p_svc = LLMProviderService(self.db)
+                        p = p_svc.get_provider(job.provider_id)
+                        if p:
+                            provider_model = p.default_model or "gpt-4o-mini"
+                    except Exception:
+                        provider_model = None
+
                 # Check token budget for this batch to determine safe max_tokens
                 token_budget = estimate_token_budget(
                     source_rows=rows_for_llm,
@@ -849,8 +1071,7 @@ class TranslationExecutor:
                     context_columns=None,  # Context is embedded in dict, not separate column
                     dictionary_entries=dict_matches,
                     batch_size=len(rows_for_llm),
-                    context_window=DEFAULT_CONTEXT_WINDOW,
-                    max_output_tokens=DEFAULT_MAX_OUTPUT_TOKENS,
+                    provider_info=provider_model,
                 )
                 if token_budget["warning"]:
                     logger.explore("Token budget warning for batch", {
@@ -1144,6 +1365,7 @@ class TranslationExecutor:
         dict_matches: list[dict[str, Any]],
         batch_id: str,
         max_tokens: int = 8192,
+        _recursion_depth: int = 0,
     ) -> dict[str, int]:
         with belief_scope("TranslationExecutor._call_llm_for_batch"):
             # Build dictionary section using ContextAwarePromptBuilder
@@ -1201,9 +1423,10 @@ class TranslationExecutor:
             last_error = None
             retries = 0
 
+            finish_reason: str | None = None
             for attempt in range(1, MAX_RETRIES_PER_BATCH + 1):
                 try:
-                    llm_response = self._call_llm(job, prompt, max_tokens=max_tokens)
+                    llm_response, finish_reason = self._call_llm(job, prompt, max_tokens=max_tokens)
                     break
                 except Exception as e:
                     last_error = str(e)
@@ -1240,6 +1463,57 @@ class TranslationExecutor:
                     self.db.add(record)
                 return {"successful": 0, "failed": len(batch_rows), "skipped": 0, "retries": retries}
 
+            # ── Truncation detection: finish_reason="length" → split batch ──
+            if finish_reason == "length" and len(batch_rows) >= 2 and run_id:
+                if _recursion_depth < MAX_RETRIES_PER_BATCH:
+                    mid = len(batch_rows) // 2
+                    logger.explore("LLM output truncated — splitting batch", {
+                        "batch_id": batch_id,
+                        "batch_size": len(batch_rows),
+                        "split_at": mid,
+                        "recursion_depth": _recursion_depth,
+                        "finish_reason": finish_reason,
+                    })
+                    left_result = self._call_llm_for_batch(
+                        job=job,
+                        run_id=run_id,
+                        batch_rows=batch_rows[:mid],
+                        dict_matches=dict_matches,
+                        batch_id=batch_id + "_L",
+                        max_tokens=max_tokens,
+                        _recursion_depth=_recursion_depth + 1,
+                    )
+                    right_result = self._call_llm_for_batch(
+                        job=job,
+                        run_id=run_id,
+                        batch_rows=batch_rows[mid:],
+                        dict_matches=dict_matches,
+                        batch_id=batch_id + "_R",
+                        max_tokens=max_tokens,
+                        _recursion_depth=_recursion_depth + 1,
+                    )
+                    merged = {
+                        "successful": left_result["successful"] + right_result["successful"],
+                        "failed": left_result["failed"] + right_result["failed"],
+                        "skipped": left_result["skipped"] + right_result["skipped"],
+                        "retries": retries + left_result.get("retries", 0) + right_result.get("retries", 0),
+                    }
+                    logger.reason("Truncation resolved by batch splitting", {
+                        "batch_id": batch_id,
+                        "left_successful": left_result["successful"],
+                        "right_successful": right_result["successful"],
+                        "left_size": len(batch_rows[:mid]),
+                        "right_size": len(batch_rows[mid:]),
+                    })
+                    return merged
+                else:
+                    logger.explore("Truncation recursion depth exceeded — accepting truncated output", {
+                        "batch_id": batch_id,
+                        "recursion_depth": _recursion_depth,
+                        "max_depth": MAX_RETRIES_PER_BATCH,
+                    })
+                    # Fall through to parse truncated response
+
             # Parse LLM response (multi-language aware)
             try:
                 translations = self._parse_llm_response(llm_response, len(batch_rows), target_languages=target_languages)
@@ -1412,7 +1686,7 @@ class TranslationExecutor:
     # @PRE: job has valid provider_id.
     # @POST: Returns raw LLM response string.
     # @SIDE_EFFECT: HTTP call to LLM provider.
-    def _call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> str:
+    def _call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> tuple[str, str | None]:
         with belief_scope("TranslationExecutor._call_llm"):
             if not job.provider_id:
                 raise ValueError("Job has no LLM provider configured")
@@ -1432,7 +1706,7 @@ class TranslationExecutor:
             disable_reasoning = getattr(job, 'disable_reasoning', False)
 
             if provider_type in ("openai", "openai_compatible", "openrouter", "kilo", "litellm"):
-                response_text = self._call_openai_compatible(
+                response_text, finish_reason = self._call_openai_compatible(
                     base_url=provider.base_url,
                     api_key=api_key,
                     model=model,
@@ -1441,7 +1715,7 @@ class TranslationExecutor:
                     max_tokens=max_tokens,
                     disable_reasoning=disable_reasoning,
                 )
-                return response_text
+                return response_text, finish_reason
             else:
                 raise ValueError(f"Unsupported provider type '{provider_type}'")
     # endregion _call_llm
@@ -1460,8 +1734,10 @@ class TranslationExecutor:
         provider_type: str = "openai",
         max_tokens: int = 8192,
         disable_reasoning: bool = False,
-    ) -> str:
+    ) -> tuple[str, str | None]:
         with belief_scope("TranslationExecutor._call_openai_compatible"):
+            if not base_url:
+                raise ValueError("LLM provider has no base_url configured")
             import requests as http_requests
 
             url = f"{base_url.rstrip('/')}/chat/completions"
@@ -1489,8 +1765,9 @@ class TranslationExecutor:
             if disable_reasoning:
                 if provider_type not in ("kilo", "openrouter", "litellm"):
                     payload["reasoning_effort"] = "none"
-                    payload["extra_body"] = {"reasoning_effort": "none"}
                 payload.pop("response_format", None)
+                # Use caller-provided max_tokens instead of hardcoded 8192
+                payload["max_tokens"] = max_tokens
                 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."}
 
             logger.reason(
@@ -1500,9 +1777,39 @@ class TranslationExecutor:
                 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 ""):
+            # ── Handle rate limiting with Retry-After header ──
+            import time as _time
+            _max_retry_429 = 3
+            _retry_count_429 = 0
+            while _retry_count_429 < _max_retry_429:
+                response = http_requests.post(url, headers=headers, json=payload, timeout=180)
+                if response.status_code == 429:
+                    _retry_count_429 += 1
+                    retry_after = response.headers.get("Retry-After")
+                    if retry_after:
+                        try:
+                            wait = int(retry_after)
+                        except (ValueError, TypeError):
+                            wait = 2 ** _retry_count_429
+                    else:
+                        wait = 2 ** _retry_count_429
+                    logger.explore(
+                        f"Rate limited (429), retry {_retry_count_429}/{_max_retry_429} "
+                        f"after {wait}s",
+                        extra={"src": "executor", "retry_after": retry_after, "wait": wait},
+                    )
+                    _time.sleep(wait)
+                    if _retry_count_429 >= _max_retry_429:
+                        break
+                else:
+                    break
+
+            _response_format_error_patterns = ("response_format", "structured_outputs", "structured", "json_object")
+            if (
+                not response.ok
+                and response.status_code == 400
+                and any(p in (response.text or "").lower() for p in _response_format_error_patterns)
+            ):
                 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)
@@ -1520,10 +1827,22 @@ 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") or "none"
-            msg = choices[0].get("message") or {}
+            try:
+                finish_reason = choices[0].get("finish_reason") or "none"
+                msg = choices[0].get("message") or {}
+            except (TypeError, AttributeError) as e:
+                logger.explore("TypeError processing LLM response choices", extra={
+                    "src": "executor_diag",
+                    "error": str(e),
+                    "choices_0_type": type(choices[0]).__name__ if choices else "N/A",
+                    "choices_0_repr": repr(choices[0])[:2000] if choices else "N/A",
+                    "data_type": type(data).__name__,
+                    "data_preview": str(data)[:2000],
+                })
+                raise ValueError(f"LLM response processing failed: {e}")
+
             # Handle model refusal (content is empty/null, refusal field has reason)
-            refusal = msg.get("refusal")
+            refusal = msg.get("refusal") if isinstance(msg, dict) else None
             if refusal:
                 logger.explore("LLM refused to respond", extra={
                     "src": "executor",
@@ -1531,13 +1850,15 @@ class TranslationExecutor:
                     "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())}")
+            content = msg.get("content") if isinstance(msg, dict) else ""
+            if not content and isinstance(msg, dict):
+                content = msg.get("content") or ""
+            logger.reason(f"LLM response finish_reason={finish_reason} content_len={len(content)} msg_keys={list(msg.keys()) if isinstance(msg, dict) else []}")
             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]})
+                logger.explore("LLM returned empty content", extra={"src": "executor", "finish_reason": finish_reason, "msg_keys": list(msg.keys()) if isinstance(msg, dict) else [], "response_preview": str(data)[:2000]})
                 raise ValueError("LLM returned empty content")
 
-            return content
+            return content, finish_reason
     # endregion _call_openai_compatible
 
     # region _parse_llm_response [TYPE Function]
@@ -1568,7 +1889,7 @@ class TranslationExecutor:
                 
                 # If finish_reason=length, try to recover complete rows from truncated JSON
                 logger.explore("LLM truncated, trying partial row recovery", extra={"src": "executor", "finish_reason": finish_reason, "response_length": len(response_text)})
-                rows_match = re.findall(r'\{\s*"row_id"\s*:\s*"\d+".*?\}\s*', response_text, re.DOTALL)
+                rows_match = re.findall(r'\{\s*"row_id"\s*:\s*(?:\d+|"\d+").*?\}\s*', response_text, re.DOTALL)
                 if rows_match:
                     partial_rows = []
                     for row_text in rows_match:
diff --git a/backend/src/plugins/translate/preview.py b/backend/src/plugins/translate/preview.py
index fac4ce08..bc39eddc 100644
--- a/backend/src/plugins/translate/preview.py
+++ b/backend/src/plugins/translate/preview.py
@@ -37,7 +37,7 @@ from ...models.translate import (
 )
 from ...services.llm_prompt_templates import render_prompt
 from ...services.llm_provider import LLMProviderService
-from ._token_budget import DEFAULT_CONTEXT_WINDOW, DEFAULT_MAX_OUTPUT_TOKENS, estimate_token_budget
+from ._token_budget import DEFAULT_CONTEXT_WINDOW, estimate_token_budget
 from .dictionary import DictionaryManager
 
 # #region DEFAULT_EXECUTION_PROMPT_TEMPLATE [TYPE Constant]
@@ -95,8 +95,8 @@ DEFAULT_PREVIEW_PROMPT_TEMPLATE: str = (
 class TokenEstimator:
     """Estimate token counts and costs for LLM operations."""
 
-    CHARS_PER_TOKEN_ESTIMATE: float = 4.0
-    OUTPUT_TOKENS_PER_ROW_ESTIMATE: int = 50
+    CHARS_PER_TOKEN_ESTIMATE: float = 2.2
+    OUTPUT_TOKENS_PER_ROW_ESTIMATE: int = 120
     MULTI_LANG_FACTOR: float = 1.2  # Overhead for multi-language in one call
     TOKEN_COST_PER_1K: float = 0.002  # Default cost per 1K tokens
     COST_WARNING_THRESHOLD: int = 30  # Show warning above this sample size
@@ -229,15 +229,25 @@ class TranslationPreview:
                     f"val='{first_row.get(job.translation_column, '')}'"
                 )
 
-            # 3b. Check token budget and auto-reduce sample size if needed
+            # 3b. Resolve provider model for provider-aware token budget
+            provider_model = None
+            if job.provider_id:
+                try:
+                    provider_svc = LLMProviderService(self.db)
+                    provider = provider_svc.get_provider(job.provider_id)
+                    if provider:
+                        provider_model = provider.default_model or "gpt-4o-mini"
+                except Exception:
+                    provider_model = None
+
+            # 3c. Check token budget and auto-reduce sample size if needed
             token_budget = estimate_token_budget(
                 source_rows=source_rows,
                 target_languages=target_languages,
                 source_column=job.translation_column,
                 context_columns=job.context_columns,
                 batch_size=actual_row_count,
-                context_window=DEFAULT_CONTEXT_WINDOW,
-                max_output_tokens=DEFAULT_MAX_OUTPUT_TOKENS,
+                provider_info=provider_model,
             )
             if token_budget["warning"]:
                 logger.explore("Token budget warning", {
@@ -265,8 +275,7 @@ class TranslationPreview:
                     source_column=job.translation_column,
                     context_columns=job.context_columns,
                     batch_size=actual_row_count,
-                    context_window=DEFAULT_CONTEXT_WINDOW,
-                    max_output_tokens=DEFAULT_MAX_OUTPUT_TOKENS,
+                    provider_info=provider_model,
                 )
 
             # 4. Build prompt context from rows
@@ -1020,6 +1029,8 @@ class TranslationPreview:
         disable_reasoning: bool = False,
     ) -> str:
         with belief_scope("TranslationPreview._call_openai_compatible"):
+            if not base_url:
+                raise ValueError("LLM provider has no base_url configured")
             import requests as http_requests
 
             url = f"{base_url.rstrip('/')}/chat/completions"
@@ -1047,10 +1058,10 @@ class TranslationPreview:
                 # Kilo/OpenRouter/LiteLLM reject reasoning_effort — only use for native OpenAI-compatible
                 if provider_type not in ("kilo", "openrouter", "litellm"):
                     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
+                # Use caller-provided max_tokens instead of hardcoded 8192
+                # This ensures multi-language batches with large output get enough token budget.
+                payload["max_tokens"] = max_tokens
                 # Universal instruction — all models understand "respond directly without reasoning"
                 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."
                 payload["messages"][0] = {"role": "system", "content": system_content}
@@ -1063,9 +1074,39 @@ class TranslationPreview:
                 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=600)
-            if not response.ok and response.status_code == 400 and "structured_outputs is not supported" in (response.text or ""):
+            # ── Handle rate limiting with Retry-After header ──
+            import time as _time
+            _max_retry_429 = 3
+            _retry_count_429 = 0
+            while _retry_count_429 < _max_retry_429:
+                response = http_requests.post(url, headers=headers, json=payload, timeout=600)
+                if response.status_code == 429:
+                    _retry_count_429 += 1
+                    retry_after = response.headers.get("Retry-After")
+                    if retry_after:
+                        try:
+                            wait = int(retry_after)
+                        except (ValueError, TypeError):
+                            wait = 2 ** _retry_count_429
+                    else:
+                        wait = 2 ** _retry_count_429
+                    logger.explore(
+                        f"Rate limited (429), retry {_retry_count_429}/{_max_retry_429} "
+                        f"after {wait}s",
+                        extra={"src": "preview", "retry_after": retry_after, "wait": wait},
+                    )
+                    _time.sleep(wait)
+                    if _retry_count_429 >= _max_retry_429:
+                        break
+                else:
+                    break
+
+            _response_format_error_patterns = ("response_format", "structured_outputs", "structured", "json_object")
+            if (
+                not response.ok
+                and response.status_code == 400
+                and any(p in (response.text or "").lower() for p in _response_format_error_patterns)
+            ):
                 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)
@@ -1146,7 +1187,7 @@ class TranslationPreview:
                 
                 # If finish_reason=length, try to recover complete rows from truncated JSON
                 logger.explore("LLM truncated, trying partial row recovery", extra={"src": "preview", "finish_reason": finish_reason, "response_length": len(response_text)})
-                rows_match = re.findall(r'\{\s*"row_id"\s*:\s*"\d+".*?\}\s*', response_text, re.DOTALL)
+                rows_match = re.findall(r'\{\s*"row_id"\s*:\s*(?:\d+|"\d+").*?\}\s*', response_text, re.DOTALL)
                 if rows_match:
                     partial_rows = []
                     for row_text in rows_match:
diff --git a/backend/src/services/llm_prompt_templates.py b/backend/src/services/llm_prompt_templates.py
index 1f59ecbe..51f0a32b 100644
--- a/backend/src/services/llm_prompt_templates.py
+++ b/backend/src/services/llm_prompt_templates.py
@@ -9,6 +9,8 @@ from __future__ import annotations
 from copy import deepcopy
 from typing import Any
 
+from ..core.logger import logger
+
 # #region DEFAULT_LLM_PROMPTS [C:2] [TYPE Constant]
 # @BRIEF Default prompt templates used by documentation, dashboard validation, and git commit generation.
 DEFAULT_LLM_PROMPTS: dict[str, str] = {
@@ -181,12 +183,21 @@ def resolve_bound_provider_id(llm_settings: Any, task_key: str) -> str:
 # #region render_prompt [C:3] [TYPE Function]
 # @BRIEF Render prompt template using deterministic placeholder replacement with graceful fallback.
 # @PRE: template is a string and variables values are already stringifiable.
-# @POST: Returns rendered prompt text with known placeholders substituted.
+# @POST: Returns rendered prompt text with known placeholders substituted. Warns about unfilled placeholders.
 # @RELATION DEPENDS_ON -> LLMProviderService
 def render_prompt(template: str, variables: dict[str, Any]) -> str:
     rendered = template
     for key, value in variables.items():
         rendered = rendered.replace("{" + key + "}", str(value))
+
+    # Warn about unfilled placeholders that would be sent to LLM
+    import re
+    unfilled = re.findall(r'\{(\w+)\}', rendered)
+    if unfilled:
+        logger.warning(
+            f"[render_prompt] Unfilled placeholders in rendered prompt: {unfilled}"
+        )
+
     return rendered
 # #endregion render_prompt
 
diff --git a/backend/src/services/llm_provider.py b/backend/src/services/llm_provider.py
index 0a948383..9e4aaa9a 100644
--- a/backend/src/services/llm_provider.py
+++ b/backend/src/services/llm_provider.py
@@ -8,6 +8,7 @@ import os
 from typing import TYPE_CHECKING
 
 from cryptography.fernet import Fernet
+from cryptography.exceptions import InvalidTag
 from sqlalchemy.orm import Session
 
 from ..core.logger import belief_scope, logger
@@ -276,8 +277,23 @@ class LLMProviderService:
                     f"[get_decrypted_api_key] Decryption successful, key length: {len(decrypted_key) if decrypted_key else 0}"
                 )
                 return decrypted_key
+            except InvalidTag as e:
+                logger.error(
+                    f"[get_decrypted_api_key] Integrity check failed (InvalidTag): {e!s}. "
+                    "The encrypted API key may be corrupted or the ENCRYPTION_KEY has changed."
+                )
+                return None
+            except ValueError as e:
+                logger.error(
+                    f"[get_decrypted_api_key] Decryption format error (ValueError): {e!s}. "
+                    "The encrypted data may not be valid Fernet ciphertext."
+                )
+                return None
             except Exception as e:
-                logger.error(f"[get_decrypted_api_key] Decryption failed: {e!s}")
+                logger.error(
+                    f"[get_decrypted_api_key] Decryption failed with unexpected error "
+                    f"({type(e).__name__}): {e!s}"
+                )
                 return None
 
     # endregion get_decrypted_api_key