feat(translate): multi-language optimization (Phase 11)

- Auto-detection of source language per row via LLM (US6)
- Multi-target translation — one LLM call for N languages (US1-US3)
- Language-aware storage: TranslationLanguage, per-language stats
- Multilingual dictionaries with language-pair-aware filtering (US7)
- Inline correction on any run result + submit-to-dictionary (US8)
- Context-aware dictionary: auto-capture row context, usage notes,
  Jaccard similarity, priority flagging in LLM prompts (US8b)
- Configurable preview sample size 1-100, cost warning at >30
- Per-language history & metrics with MetricSnapshot preservation
- 36 files, +5022/-373, all specs GRACE-Poly v2.6 compliant

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

@@ -1,29 +1,37 @@
 # region DictionaryTests [TYPE Module]
-# @SEMANTICS: tests, dictionary, crud, import, filter
-# @PURPOSE: Validate DictionaryManager CRUD, import, deletion guards, and batch filtering.
+# @SEMANTICS: tests, dictionary, crud, import, filter, language_pair
+# @PURPOSE: Validate DictionaryManager CRUD, import, deletion guards, batch filtering, and language-pair support.
 # @RELATION: BINDS_TO -> [DictionaryManager:Class]
 #
 # @TEST_CONTRACT: [DictionaryManager] -> {
 #   invariants: [
 #     "Create/Read/Update/Delete dictionaries works",
-#     "Entry CRUD enforces unique source_term_normalized per dictionary",
+#     "Entry CRUD enforces unique (dictionary_id, source_term_normalized, source_language, target_language)",
 #     "Import CSV/TSV handles overwrite/keep_existing/cancel conflict modes",
 #     "Delete blocked when dictionary attached to active/scheduled jobs",
-#     "filter_for_batch returns matched entries with word-boundary awareness"
+#     "filter_for_batch returns matched entries with word-boundary awareness",
+#     "filter_for_batch supports language-pair filtering",
+#     "Language-pair-aware duplicate detection: same term with different lang pair is allowed"
 #   ]
 # }
-# @TEST_EDGE: duplicate_entry -> 409-style ValueError on repeated source_term
+# @TEST_EDGE: duplicate_entry -> 409-style ValueError on repeated (dictionary_id, source_term_norm, source_lang, target_lang)
 # @TEST_EDGE: delete_active_job -> ValueError with active/scheduled message
 # @TEST_EDGE: import_invalid_format -> ValueError for missing columns
-# @TEST_INVARIANT: unique_normalized -> verifies: [duplicate_entry]
+# @TEST_EDGE: same_term_different_lang_pair -> allowed (not duplicate)
+# @TEST_INVARIANT: unique_normalized -> verifies: [duplicate_entry, same_term_different_lang_pair allowed]
 
 
+import csv
+import io
+
 import pytest
 from sqlalchemy import create_engine, event
 from sqlalchemy.orm import Session, sessionmaker
 
 from src.models.translate import (
     Base,
+    DictionaryEntry,
+    TerminologyDictionary,
     TranslationJob,
     TranslationJobDictionary,
 )
@@ -112,8 +120,8 @@ def test_delete_dictionary(db_session: Session):
         db_session, name="To Delete",
         source_dialect="a", target_dialect="b",
     )
-    # Add an entry
-    entry = DictionaryManager.add_entry(db_session, d.id, "hello", "hola")
+    # Add an entry with language pair
+    entry = DictionaryManager.add_entry(db_session, d.id, "hello", "hola", source_language="en", target_language="es")
     assert entry.id is not None
 
     DictionaryManager.delete_dictionary(db_session, d.id)
@@ -144,15 +152,15 @@ def test_add_entry_duplicate(db_session: Session):
         db_session, name="Test",
         source_dialect="a", target_dialect="b",
     )
-    DictionaryManager.add_entry(db_session, d.id, "Hello", "Hola")
+    DictionaryManager.add_entry(db_session, d.id, "Hello", "Hola", source_language="en", target_language="es")
 
-    # Same normalized term should raise
+    # Same normalized term with same language pair should raise
     with pytest.raises(ValueError, match="already exists"):
-        DictionaryManager.add_entry(db_session, d.id, "hello", "Bonjour")
+        DictionaryManager.add_entry(db_session, d.id, "hello", "Bonjour", source_language="en", target_language="es")
 
-    # Different case, same normalized should also raise
+    # Different case, same normalized, same lang pair should also raise
     with pytest.raises(ValueError, match="already exists"):
-        DictionaryManager.add_entry(db_session, d.id, "HELLO", "Ciao")
+        DictionaryManager.add_entry(db_session, d.id, "HELLO", "Ciao", source_language="en", target_language="es")
 # endregion test_add_entry_duplicate
 
 
@@ -167,9 +175,9 @@ def test_add_entry_duplicate_per_dictionary(db_session: Session):
         db_session, name="Dict2",
         source_dialect="a", target_dialect="b",
     )
-    DictionaryManager.add_entry(db_session, d1.id, "hello", "hola")
+    DictionaryManager.add_entry(db_session, d1.id, "hello", "hola", source_language="en", target_language="es")
     # Same term in different dictionary should work
-    entry = DictionaryManager.add_entry(db_session, d2.id, "hello", "bonjour")
+    entry = DictionaryManager.add_entry(db_session, d2.id, "hello", "bonjour", source_language="en", target_language="es")
     assert entry.id is not None
 # endregion test_add_entry_duplicate_per_dictionary
 
@@ -181,13 +189,13 @@ def test_edit_entry(db_session: Session):
         db_session, name="Test",
         source_dialect="a", target_dialect="b",
     )
-    entry = DictionaryManager.add_entry(db_session, d.id, "hello", "hola")
+    entry = DictionaryManager.add_entry(db_session, d.id, "hello", "hola", source_language="en", target_language="es")
 
     updated = DictionaryManager.edit_entry(db_session, entry.id, target_term="HOLA!")
     assert updated.target_term == "HOLA!"
 
     # Edit source term to a value that's already taken should fail
-    DictionaryManager.add_entry(db_session, d.id, "world", "mundo")
+    DictionaryManager.add_entry(db_session, d.id, "world", "mundo", source_language="en", target_language="es")
     with pytest.raises(ValueError, match="already exists"):
         DictionaryManager.edit_entry(db_session, entry.id, source_term="WORLD")
 # endregion test_edit_entry
@@ -200,7 +208,7 @@ def test_delete_entry(db_session: Session):
         db_session, name="Test",
         source_dialect="a", target_dialect="b",
     )
-    entry = DictionaryManager.add_entry(db_session, d.id, "hello", "hola")
+    entry = DictionaryManager.add_entry(db_session, d.id, "hello", "hola", source_language="en", target_language="es")
     DictionaryManager.delete_entry(db_session, entry.id)
 
     entries, total = DictionaryManager.list_entries(db_session, d.id)
@@ -215,10 +223,10 @@ def test_import_csv_overwrite(db_session: Session):
         db_session, name="Test",
         source_dialect="a", target_dialect="b",
     )
-    # Pre-add an entry
-    DictionaryManager.add_entry(db_session, d.id, "hello", "hola")
+    # Pre-add an entry with language pair
+    DictionaryManager.add_entry(db_session, d.id, "hello", "hola", source_language="en", target_language="es")
 
-    csv_content = "source_term,target_term,context_notes\nhello,HELLO,\nworld,mundo,"
+    csv_content = "source_term,target_term,context_notes,source_language,target_language\nhello,HELLO,,en,es\nworld,mundo,,en,es"
     result = DictionaryManager.import_entries(
         db_session, d.id, csv_content,
         delimiter=",", on_conflict="overwrite",
@@ -242,9 +250,9 @@ def test_import_csv_keep_existing(db_session: Session):
         db_session, name="Test",
         source_dialect="a", target_dialect="b",
     )
-    DictionaryManager.add_entry(db_session, d.id, "hello", "hola")
+    DictionaryManager.add_entry(db_session, d.id, "hello", "hola", source_language="en", target_language="es")
 
-    csv_content = "source_term,target_term\nhello,HELLO\nworld,mundo"
+    csv_content = "source_term,target_term,source_language,target_language\nhello,HELLO,en,es\nworld,mundo,en,es"
     result = DictionaryManager.import_entries(
         db_session, d.id, csv_content,
         delimiter=",", on_conflict="keep_existing",
@@ -267,9 +275,9 @@ def test_import_csv_cancel_on_conflict(db_session: Session):
         db_session, name="Test",
         source_dialect="a", target_dialect="b",
     )
-    DictionaryManager.add_entry(db_session, d.id, "hello", "hola")
+    DictionaryManager.add_entry(db_session, d.id, "hello", "hola", source_language="en", target_language="es")
 
-    csv_content = "source_term,target_term\nhello,HELLO\nworld,mundo"
+    csv_content = "source_term,target_term,source_language,target_language\nhello,HELLO,en,es\nworld,mundo,en,es"
     result = DictionaryManager.import_entries(
         db_session, d.id, csv_content,
         delimiter=",", on_conflict="cancel",
@@ -289,7 +297,7 @@ def test_import_tsv(db_session: Session):
         db_session, name="Test",
         source_dialect="a", target_dialect="b",
     )
-    tsv_content = "source_term\ttarget_term\nhello\thola\nworld\tmundo"
+    tsv_content = "source_term\ttarget_term\tsource_language\ttarget_language\nhello\thola\ten\tes\nworld\tmundo\ten\tes"
     result = DictionaryManager.import_entries(
         db_session, d.id, tsv_content,
         delimiter="\t", on_conflict="overwrite",
@@ -339,9 +347,9 @@ def test_import_preview(db_session: Session):
         db_session, name="Test",
         source_dialect="a", target_dialect="b",
     )
-    DictionaryManager.add_entry(db_session, d.id, "hello", "hola")
+    DictionaryManager.add_entry(db_session, d.id, "hello", "hola", source_language="en", target_language="es")
 
-    csv_content = "source_term,target_term\nhello,HELLO\nworld,mundo"
+    csv_content = "source_term,target_term,source_language,target_language\nhello,HELLO,en,es\nworld,mundo,en,es"
     result = DictionaryManager.import_entries(
         db_session, d.id, csv_content,
         delimiter=",", on_conflict="overwrite",
@@ -446,9 +454,9 @@ def test_filter_for_batch_matches(db_session: Session):
         db_session, name="Test Dict",
         source_dialect="a", target_dialect="b",
     )
-    DictionaryManager.add_entry(db_session, d.id, "hello", "hola")
-    DictionaryManager.add_entry(db_session, d.id, "world", "mundo")
-    DictionaryManager.add_entry(db_session, d.id, "foo", "bar")
+    DictionaryManager.add_entry(db_session, d.id, "hello", "hola", source_language="en", target_language="es")
+    DictionaryManager.add_entry(db_session, d.id, "world", "mundo", source_language="en", target_language="es")
+    DictionaryManager.add_entry(db_session, d.id, "foo", "bar", source_language="en", target_language="es")
 
     job = TranslationJob(
         name="Test Job", source_dialect="a", target_dialect="b", status="DRAFT",
@@ -492,7 +500,7 @@ def test_filter_for_batch_case_insensitive(db_session: Session):
         db_session, name="Test Dict",
         source_dialect="a", target_dialect="b",
     )
-    DictionaryManager.add_entry(db_session, d.id, "Hello World", "Hola Mundo")
+    DictionaryManager.add_entry(db_session, d.id, "Hello World", "Hola Mundo", source_language="en", target_language="es")
 
     job = TranslationJob(
         name="Test Job", source_dialect="a", target_dialect="b", status="DRAFT",
@@ -523,7 +531,7 @@ def test_filter_for_batch_word_boundary(db_session: Session):
         db_session, name="Test Dict",
         source_dialect="a", target_dialect="b",
     )
-    DictionaryManager.add_entry(db_session, d.id, "cat", "gato")
+    DictionaryManager.add_entry(db_session, d.id, "cat", "gato", source_language="en", target_language="es")
 
     job = TranslationJob(
         name="Test Job", source_dialect="a", target_dialect="b", status="DRAFT",
@@ -562,8 +570,8 @@ def test_filter_for_batch_multi_dictionary_priority(db_session: Session):
     d2 = DictionaryManager.create_dictionary(
         db_session, name="Priority2", source_dialect="a", target_dialect="b",
     )
-    DictionaryManager.add_entry(db_session, d1.id, "hello", "hola")
-    DictionaryManager.add_entry(db_session, d2.id, "hello", "bonjour")
+    DictionaryManager.add_entry(db_session, d1.id, "hello", "hola", source_language="en", target_language="es")
+    DictionaryManager.add_entry(db_session, d2.id, "hello", "bonjour", source_language="en", target_language="es")
 
     job = TranslationJob(
         name="Test Job", source_dialect="a", target_dialect="b", status="DRAFT",
@@ -625,8 +633,8 @@ def test_clear_entries(db_session: Session):
     d = DictionaryManager.create_dictionary(
         db_session, name="Test", source_dialect="a", target_dialect="b",
     )
-    DictionaryManager.add_entry(db_session, d.id, "hello", "hola")
-    DictionaryManager.add_entry(db_session, d.id, "world", "mundo")
+    DictionaryManager.add_entry(db_session, d.id, "hello", "hola", source_language="en", target_language="es")
+    DictionaryManager.add_entry(db_session, d.id, "world", "mundo", source_language="en", target_language="es")
 
     deleted = DictionaryManager.clear_entries(db_session, d.id)
     assert deleted == 2
@@ -634,4 +642,558 @@ def test_clear_entries(db_session: Session):
     entries, total = DictionaryManager.list_entries(db_session, d.id)
     assert total == 0
 # endregion test_clear_entries
+
+# region test_add_entry_with_language_pair [TYPE Function]
+# @PURPOSE: Verify creating entry with source_language and target_language stores correctly.
+def test_add_entry_with_language_pair(db_session: Session):
+    d = DictionaryManager.create_dictionary(
+        db_session, name="Lang Test",
+        source_dialect="a", target_dialect="b",
+    )
+    entry = DictionaryManager.add_entry(
+        db_session, d.id, "hello", "привет",
+        source_language="en", target_language="ru",
+    )
+    assert entry.source_language == "en"
+    assert entry.target_language == "ru"
+    assert entry.source_term == "hello"
+
+    # Read back
+    entries, total = DictionaryManager.list_entries(db_session, d.id)
+    assert total == 1
+    assert entries[0].source_language == "en"
+    assert entries[0].target_language == "ru"
+# endregion test_add_entry_with_language_pair
+
+
+# region test_duplicate_same_language_pair [TYPE Function]
+# @PURPOSE: Verify duplicate entry with same (dictionary_id, source_term_norm, source_lang, target_lang) raises conflict.
+def test_duplicate_same_language_pair(db_session: Session):
+    d = DictionaryManager.create_dictionary(
+        db_session, name="Dup Test",
+        source_dialect="a", target_dialect="b",
+    )
+    DictionaryManager.add_entry(
+        db_session, d.id, "hello", "привет",
+        source_language="en", target_language="ru",
+    )
+    # Same term with same language pair should raise
+    with pytest.raises(ValueError, match="already exists"):
+        DictionaryManager.add_entry(
+            db_session, d.id, "hello", "hallo",
+            source_language="en", target_language="ru",
+        )
+# endregion test_duplicate_same_language_pair
+
+
+# region test_same_term_different_language_pair [TYPE Function]
+# @PURPOSE: Verify same source_term with different language pair is allowed (not duplicate).
+def test_same_term_different_language_pair(db_session: Session):
+    d = DictionaryManager.create_dictionary(
+        db_session, name="Multi Lang",
+        source_dialect="a", target_dialect="b",
+    )
+    entry1 = DictionaryManager.add_entry(
+        db_session, d.id, "hello", "привет",
+        source_language="en", target_language="ru",
+    )
+    entry2 = DictionaryManager.add_entry(
+        db_session, d.id, "hello", "hallo",
+        source_language="en", target_language="de",
+    )
+    assert entry1.id != entry2.id
+    assert entry1.target_language == "ru"
+    assert entry2.target_language == "de"
+
+    entries, total = DictionaryManager.list_entries(db_session, d.id)
+    assert total == 2
+# endregion test_same_term_different_language_pair
+
+
+# region test_filter_for_batch_with_language_pair [TYPE Function]
+# @PURPOSE: Verify filter_for_batch with source_language and target_language filters entries correctly.
+def test_filter_for_batch_with_language_pair(db_session: Session):
+    d = DictionaryManager.create_dictionary(
+        db_session, name="Lang Filter",
+        source_dialect="a", target_dialect="b",
+    )
+    DictionaryManager.add_entry(db_session, d.id, "hello", "привет", source_language="en", target_language="ru")
+    DictionaryManager.add_entry(db_session, d.id, "world", "мир", source_language="en", target_language="ru")
+    DictionaryManager.add_entry(db_session, d.id, "hello", "hallo", source_language="en", target_language="de")
+
+    job = TranslationJob(
+        name="Lang Job", source_dialect="a", target_dialect="b", status="DRAFT",
+    )
+    db_session.add(job)
+    db_session.flush()
+
+    link = TranslationJobDictionary(job_id=job.id, dictionary_id=d.id)
+    db_session.add(link)
+    db_session.commit()
+
+    # Filter for en → ru: should get 2 entries (hello/privet, world/mir)
+    result = DictionaryManager.filter_for_batch(
+        db_session, ["hello world"], job.id,
+        source_language="en", target_language="ru",
+    )
+    assert len(result) == 2
+    assert all(m["target_language"] == "ru" for m in result)
+
+    # Filter for en → de: should get only the hello/hallo entry
+    result2 = DictionaryManager.filter_for_batch(
+        db_session, ["hello"], job.id,
+        source_language="en", target_language="de",
+    )
+    assert len(result2) == 1
+    assert result2[0]["target_language"] == "de"
+    assert result2[0]["target_term"] == "hallo"
+# endregion test_filter_for_batch_with_language_pair
+
+
+# region test_filter_for_batch_target_language_only [TYPE Function]
+# @PURPOSE: Verify filter_for_batch with only target_language filters by target language only.
+def test_filter_for_batch_target_language_only(db_session: Session):
+    d = DictionaryManager.create_dictionary(
+        db_session, name="Target Only",
+        source_dialect="a", target_dialect="b",
+    )
+    DictionaryManager.add_entry(db_session, d.id, "hello", "привет", source_language="en", target_language="ru")
+    DictionaryManager.add_entry(db_session, d.id, "hello", "hallo", source_language="en", target_language="de")
+    DictionaryManager.add_entry(db_session, d.id, "bonjour", "hallo", source_language="fr", target_language="de")
+
+    job = TranslationJob(
+        name="Tgt Job", source_dialect="a", target_dialect="b", status="DRAFT",
+    )
+    db_session.add(job)
+    db_session.flush()
+
+    link = TranslationJobDictionary(job_id=job.id, dictionary_id=d.id)
+    db_session.add(link)
+    db_session.commit()
+
+    # Filter for target=de only: should match both hello→hallo and bonjour→hallo (from any source lang)
+    result = DictionaryManager.filter_for_batch(
+        db_session, ["hello bonjour"], job.id,
+        target_language="de",
+    )
+    assert len(result) == 2
+    assert all(m["target_language"] == "de" for m in result)
+
+    # Filter for target=ru only: should match hello→privet
+    result2 = DictionaryManager.filter_for_batch(
+        db_session, ["hello"], job.id,
+        target_language="ru",
+    )
+    assert len(result2) == 1
+    assert result2[0]["target_language"] == "ru"
+# endregion test_filter_for_batch_target_language_only
+
+
+# region test_migrate_old_entries [TYPE Function]
+# @PURPOSE: Verify migration populates language pair fields for old-style entries.
+def test_migrate_old_entries(db_session: Session):
+    # Create a dictionary with deprecated target_language set
+    d = TerminologyDictionary(
+        name="Old Dict",
+        source_dialect="a",
+        target_dialect="b",
+        target_language="ru",
+    )
+    db_session.add(d)
+    db_session.flush()
+
+    # Create old-style entries without explicit language pair (source_language=und, target_language=und)
+    entry1 = DictionaryEntry(
+        dictionary_id=d.id,
+        source_term="hello",
+        source_term_normalized="hello",
+        target_term="привет",
+        source_language="und",
+        target_language="und",
+    )
+    entry2 = DictionaryEntry(
+        dictionary_id=d.id,
+        source_term="world",
+        source_term_normalized="world",
+        target_term="мир",
+        source_language="und",
+        target_language="und",
+        origin_source_language="en",
+    )
+    db_session.add(entry1)
+    db_session.add(entry2)
+    db_session.commit()
+
+    # Run migration
+    result = DictionaryManager.migrate_old_entries(db_session)
+
+    # Verify migration counts
+    assert result["total_processed"] >= 2
+    assert result["migrated_source"] >= 1  # entry2 had origin_source_language
+    assert result["migrated_target"] >= 2  # both should get target_language from dictionary
+
+    # Verify entry1 got target_language from dictionary
+    db_session.refresh(entry1)
+    assert entry1.target_language == "ru"
+    # source_language should remain "und" (no origin_source_language)
+    assert entry1.source_language == "und"
+
+    # Verify entry2 got source from origin and target from dictionary
+    db_session.refresh(entry2)
+    assert entry2.source_language == "en"
+    assert entry2.target_language == "ru"
+# endregion test_migrate_old_entries
+
+
+# region test_export_entries [TYPE Function]
+# @PURPOSE: Verify export_entries includes language columns.
+def test_export_entries(db_session: Session):
+    d = DictionaryManager.create_dictionary(
+        db_session, name="Export Test",
+        source_dialect="a", target_dialect="b",
+    )
+    DictionaryManager.add_entry(db_session, d.id, "hello", "привет", source_language="en", target_language="ru")
+    DictionaryManager.add_entry(db_session, d.id, "world", "мир", source_language="en", target_language="ru")
+
+    csv_output = DictionaryManager.export_entries(db_session, d.id)
+    assert "source_language" in csv_output
+    assert "target_language" in csv_output
+    assert "context_data" in csv_output
+    assert "usage_notes" in csv_output
+    assert "en" in csv_output
+    assert "ru" in csv_output
+
+    # Verify it's valid CSV
+    reader = csv.DictReader(io.StringIO(csv_output))
+    rows = list(reader)
+    assert len(rows) == 2
+    assert rows[0]["source_language"] == "en"
+    assert rows[0]["target_language"] == "ru"
+# endregion test_export_entries
+
+
+# region test_import_with_default_language [TYPE Function]
+# @PURPOSE: Verify import uses default_source_language and default_target_language when columns missing.
+def test_import_with_default_language(db_session: Session):
+    d = DictionaryManager.create_dictionary(
+        db_session, name="Default Lang Import",
+        source_dialect="a", target_dialect="b",
+    )
+    # CSV without language columns
+    csv_content = "source_term,target_term\nhello,привет\nworld,мир"
+    result = DictionaryManager.import_entries(
+        db_session, d.id, csv_content,
+        delimiter=",", on_conflict="overwrite",
+        default_source_language="en",
+        default_target_language="ru",
+    )
+    assert result["created"] == 2
+
+    entries, _ = DictionaryManager.list_entries(db_session, d.id)
+    for e in entries:
+        assert e.source_language == "en"
+        assert e.target_language == "ru"
+# endregion test_import_with_default_language
+
+
+# region test_context_capture_in_correction [TYPE Function]
+# @PURPOSE: Verify context_data is auto-captured when submitting a correction with source row context.
+def test_context_capture_in_correction(db_session: Session):
+    """Test T132: Context auto-capture when submitting a correction."""
+    from src.plugins.translate.service import InlineCorrectionService
+    from src.models.translate import TranslationRecord, TranslationRun, TranslationBatch, TranslationLanguage
+
+    d = DictionaryManager.create_dictionary(
+        db_session, name="Test Dict",
+        source_dialect="a", target_dialect="b",
+    )
+    job = TranslationJob(
+        name="Ctx Job", source_dialect="a", target_dialect="b",
+        status="DRAFT",
+    )
+    db_session.add(job)
+    db_session.flush()
+
+    run = TranslationRun(
+        job_id=job.id, status="COMPLETED",
+    )
+    db_session.add(run)
+    db_session.flush()
+
+    batch = TranslationBatch(
+        run_id=run.id, batch_index=0, status="COMPLETED",
+    )
+    db_session.add(batch)
+    db_session.flush()
+
+    # Create a TranslationRecord with source_data (context columns)
+    record = TranslationRecord(
+        batch_id=batch.id,
+        run_id=run.id,
+        source_sql="hello world",
+        source_object_type="table_row",
+        source_object_name="Row 0",
+        source_object_id="0",
+        source_data={"schema": "public", "table": "users", "column": "name"},
+        status="SUCCESS",
+    )
+    db_session.add(record)
+    db_session.flush()
+
+    # Create language entries for test languages
+    for lang_code in ("ru", "de", "fr", "it"):
+        tl = TranslationLanguage(
+            record_id=record.id,
+            language_code=lang_code,
+            source_language_detected="en",
+            translated_value="translated",
+            final_value="translated",
+            status="translated",
+        )
+        db_session.add(tl)
+    db_session.commit()
+
+    # Submit correction to dictionary
+    result = InlineCorrectionService.submit_correction_to_dict(
+        db=db_session,
+        record_id=record.id,
+        language_code="ru",
+        dictionary_id=d.id,
+        corrected_value="привет мир (исправлено)",
+        current_user="test_user",
+    )
+
+    assert result["action"] == "created"
+    assert result["entry_id"] is not None
+
+    # Verify context was auto-captured
+    entry = db_session.query(DictionaryEntry).filter(DictionaryEntry.id == result["entry_id"]).first()
+    assert entry is not None
+    assert entry.has_context is True
+    assert entry.context_source == "auto"
+    assert entry.context_data is not None
+    # source_data should contain the row's context columns
+    assert "source_data" in entry.context_data
+    assert entry.context_data["source_object_type"] == "table_row"
+    assert entry.context_data["source_object_name"] == "Row 0"
+
+    # Test context_data_override (use different language to avoid conflict)
+    result2 = InlineCorrectionService.submit_correction_to_dict(
+        db=db_session,
+        record_id=record.id,
+        language_code="de",  # different target language avoids conflict
+        dictionary_id=d.id,
+        corrected_value="another correction",
+        current_user="test_user",
+        context_data_override={"custom_key": "custom_value"},
+    )
+    assert result2["action"] == "created"
+    entry2 = db_session.query(DictionaryEntry).filter(DictionaryEntry.id == result2["entry_id"]).first()
+    assert entry2.context_data == {"custom_key": "custom_value"}
+    assert entry2.context_source == "manual"
+
+    # Test keep_context=False
+    result3 = InlineCorrectionService.submit_correction_to_dict(
+        db=db_session,
+        record_id=record.id,
+        language_code="fr",  # different target language avoids conflict
+        dictionary_id=d.id,
+        corrected_value="no context",
+        current_user="test_user",
+        keep_context=False,
+    )
+    assert result3["action"] == "created"
+    entry3 = db_session.query(DictionaryEntry).filter(DictionaryEntry.id == result3["entry_id"]).first()
+    assert entry3.context_data is None
+    assert entry3.has_context is False
+    assert entry3.context_source == "manual"
+
+    # Test usage_notes
+    result4 = InlineCorrectionService.submit_correction_to_dict(
+        db=db_session,
+        record_id=record.id,
+        language_code="it",  # different target language avoids conflict
+        dictionary_id=d.id,
+        corrected_value="with notes",
+        current_user="test_user",
+        usage_notes="Use this for user-facing columns only",
+    )
+    assert result4["action"] == "created"
+    entry4 = db_session.query(DictionaryEntry).filter(DictionaryEntry.id == result4["entry_id"]).first()
+    assert entry4.usage_notes == "Use this for user-facing columns only"
+# endregion test_context_capture_in_correction
+
+
+# region test_jaccard_similarity [TYPE Function]
+# @PURPOSE: Verify Jaccard similarity computation in ContextAwarePromptBuilder.
+def test_jaccard_similarity():
+    """Test T132: Jaccard similarity = 1.0 for identical, 0.0 for disjoint."""
+    from src.plugins.translate.prompt_builder import ContextAwarePromptBuilder
+
+    builder = ContextAwarePromptBuilder()
+
+    # Identical contexts
+    ctx1 = {"schema": "public", "table": "users"}
+    ctx2 = {"schema": "public", "table": "users"}
+    sim = builder.compute_context_similarity(ctx1, ctx2)
+    assert sim == 1.0, f"Expected 1.0, got {sim}"
+
+    # Disjoint contexts
+    ctx3 = {"schema": "private"}
+    sim = builder.compute_context_similarity(ctx1, ctx3)
+    assert sim == 0.0, f"Expected 0.0, got {sim}"
+
+    # Partial overlap
+    ctx4 = {"schema": "public", "table": "orders"}
+    sim = builder.compute_context_similarity(ctx1, ctx4)
+    # intersection: {"public"} union: {"public", "users", "orders"} => 1/3 ≈ 0.33
+    assert 0.33 <= sim <= 0.34, f"Expected ~0.33, got {sim}"
+
+    # Empty/missing contexts
+    assert builder.compute_context_similarity(None, ctx1) == 0.0
+    assert builder.compute_context_similarity(ctx1, None) == 0.0
+    assert builder.compute_context_similarity({}, ctx1) == 0.0
+
+    # Case-insensitive matching (same keys, different case)
+    ctx5 = {"schema": "PUBLIC", "table": "USERS"}
+    sim = builder.compute_context_similarity(ctx1, ctx5)
+    assert sim == 1.0, f"Expected 1.0 for case-insensitive, got {sim}"
+# endregion test_jaccard_similarity
+
+
+# region test_context_truncation [TYPE Function]
+# @PURPOSE: Verify context string truncation at ~2000 chars.
+def test_context_truncation():
+    """Test T132: Context string truncation in render_entry."""
+    from src.plugins.translate.prompt_builder import ContextAwarePromptBuilder
+
+    builder = ContextAwarePromptBuilder()
+
+    # Build context data that would produce a very long context string
+    long_val = "x" * 500
+    long_context = {f"key{i}": long_val for i in range(10)}
+
+    # Create an entry-like dict with this context
+    entry = {
+        "source_term": "hello",
+        "target_term": "world",
+        "has_context": True,
+        "context_data": long_context,
+        "usage_notes": None,
+    }
+
+    rendered = builder.render_entry(entry, priority=False)
+    # The context part should be truncated
+    assert len(rendered) < 2200, f"Rendered length {len(rendered)} exceeds expected max"
+    assert "...[truncated]" in rendered, "Expected truncation marker in output"
+
+    # Short context should not be truncated
+    short_entry = {
+        "source_term": "hello",
+        "target_term": "world",
+        "has_context": True,
+        "context_data": {"table": "users"},
+        "usage_notes": None,
+    }
+    short_rendered = builder.render_entry(short_entry, priority=False)
+    assert "...[truncated]" not in short_rendered
+    assert "(context: table=users)" in short_rendered
+
+    # Priority prefix
+    priority_rendered = builder.render_entry(short_entry, priority=True)
+    assert priority_rendered.startswith("# PRIORITY (context match)")
+
+    # Usage notes
+    notes_entry = {
+        "source_term": "hello",
+        "target_term": "world",
+        "has_context": False,
+        "context_data": None,
+        "usage_notes": "Use for admin panels",
+    }
+    notes_rendered = builder.render_entry(notes_entry, priority=False)
+    assert "# Usage: Use for admin panels" in notes_rendered
+# endregion test_context_truncation
+
+
+# region test_render_entry_dict_and_object [TYPE Function]
+# @PURPOSE: Verify render_entry handles both dicts and objects.
+def test_render_entry_dict_and_object():
+    """Test T132: render_entry accepts dict or object with same results."""
+    from src.plugins.translate.prompt_builder import ContextAwarePromptBuilder
+
+    builder = ContextAwarePromptBuilder()
+
+    # Dict entry
+    dict_entry = {
+        "source_term": "hello",
+        "target_term": "hola",
+        "has_context": False,
+        "context_data": None,
+        "usage_notes": None,
+    }
+    dict_result = builder.render_entry(dict_entry)
+
+    # Object-like entry using a simple class
+    class FakeEntry:
+        source_term = "hello"
+        target_term = "hola"
+        has_context = False
+        context_data = None
+        usage_notes = None
+
+    obj_result = builder.render_entry(FakeEntry())
+    assert dict_result == obj_result
+    assert dict_result == '"hello" -> "hola"'
+# endregion test_render_entry_dict_and_object
+
+
+# region test_build_context_entries_prioritization [TYPE Function]
+# @PURPOSE: Verify build_context_entries sorts priority entries first.
+def test_build_context_entries_prioritization():
+    """Test T132: build_context_entries sorts priority entries before non-priority."""
+    from src.plugins.translate.prompt_builder import ContextAwarePromptBuilder
+
+    builder = ContextAwarePromptBuilder()
+
+    entries = [
+        {
+            "source_term": "high_match",
+            "target_term": "alta_coincidencia",
+            "has_context": True,
+            "context_data": {"schema": "public", "table": "users"},
+            "usage_notes": None,
+        },
+        {
+            "source_term": "no_match",
+            "target_term": "sin_coincidencia",
+            "has_context": True,
+            "context_data": {"schema": "private", "table": "config"},
+            "usage_notes": None,
+        },
+        {
+            "source_term": "no_context",
+            "target_term": "sin_contexto",
+            "has_context": False,
+            "context_data": None,
+            "usage_notes": None,
+        },
+    ]
+
+    # Row context that matches the first entry
+    row_context = {"schema": "public", "table": "users"}
+
+    rendered = builder.build_context_entries(entries, row_context)
+
+    # The first entry should be priority (similarity=1.0 >= 0.5)
+    assert len(rendered) == 3
+    assert rendered[0].startswith("# PRIORITY (context match)"), f"Expected priority first, got: {rendered[0]}"
+    assert '"high_match"' in rendered[0]
+
+    # The non-priority entries should not have priority prefix
+    assert not rendered[1].startswith("# PRIORITY")
+    assert not rendered[2].startswith("# PRIORITY")
+# endregion test_build_context_entries_prioritization
+
+
 # endregion DictionaryTests

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

@@ -11,6 +11,7 @@
 # @TEST_EDGE: invalid_run_status -> raises ValueError
 # @TEST_EDGE: executor_failure -> run is marked FAILED
 
+import json
 from datetime import UTC, datetime
 from unittest.mock import MagicMock, patch
 
@@ -18,11 +19,14 @@ import pytest
 
 from src.models.translate import (
     TranslationJob,
+    TranslationLanguage,
     TranslationPreviewSession,
     TranslationRecord,
     TranslationRun,
+    TranslationRunLanguageStats,
 )
 from src.plugins.translate.events import TranslationEventLog
+from src.plugins.translate.executor import TranslationExecutor
 from src.plugins.translate.orchestrator import TranslationOrchestrator
 
 
@@ -598,4 +602,249 @@ class TestTranslationOrchestrator:
     # endregion test_get_run_records
     # endregion test_get_run_history
 # endregion TestTranslationOrchestrator
+
+
+# region TestTranslationExecutorMultiLang [TYPE Class]
+# @PURPOSE: Tests for multi-language LLM translation execution: parsing, per-language entries, source-as-reference, backward compat.
+class TestTranslationExecutorMultiLang:
+
+    # region test_parse_multi_language_response [TYPE Function]
+    # @PURPOSE: _parse_llm_response handles multi-language JSON format correctly.
+    def test_parse_multi_language_response(self) -> None:
+        response = json.dumps({
+            "rows": [
+                {"row_id": 0, "detected_source_language": "fr", "ru": "текст", "en": "text", "de": "Text"},
+                {"row_id": 1, "detected_source_language": "en", "ru": "дашборд", "en": "dashboard", "de": "Dashboard"},
+            ]
+        })
+        result = TranslationExecutor._parse_llm_response(
+            response, expected_count=2, target_languages=["ru", "en", "de"]
+        )
+        assert len(result) == 2
+        assert "0" in result
+        assert result["0"]["detected_source_language"] == "fr"
+        assert result["0"]["ru"] == "текст"
+        assert result["0"]["en"] == "text"
+        assert result["0"]["de"] == "Text"
+        assert "1" in result
+        assert result["1"]["en"] == "dashboard"
+        assert result["1"]["ru"] == "дашборд"
+    # endregion test_parse_multi_language_response
+
+    # region test_parse_legacy_format_backward_compat [TYPE Function]
+    # @PURPOSE: _parse_llm_response still handles legacy single "translation" key format.
+    def test_parse_legacy_format_backward_compat(self) -> None:
+        response = json.dumps({
+            "rows": [
+                {"row_id": 0, "translation": "текст", "detected_source_language": "fr"},
+                {"row_id": 1, "translation": "дашборд", "detected_source_language": "en"},
+            ]
+        })
+        result = TranslationExecutor._parse_llm_response(
+            response, expected_count=2, target_languages=["ru"]
+        )
+        assert len(result) == 2
+        assert result["0"]["translation"] == "текст"
+        assert result["1"]["detected_source_language"] == "en"
+    # endregion test_parse_legacy_format_backward_compat
+
+    # region test_multi_language_translation_language_entries [TYPE Function]
+    # @PURPOSE: Multi-language LLM response creates per-language TranslationLanguage entries.
+    def test_multi_language_translation_language_entries(self) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        # Mock job with multi-language target_languages
+        job = MagicMock(spec=TranslationJob)
+        job.id = "job-ml-1"
+        job.target_languages = ["ru", "en"]
+        job.target_language = None
+        job.target_dialect = "postgresql"
+        job.source_dialect = "postgresql"
+        job.translation_column = "name"
+        job.provider_id = "provider-1"
+        job.batch_size = 50
+
+        executor = TranslationExecutor(db, config_manager, "test-user")
+
+        # Patch _call_llm to return multi-language response
+        multi_lang_response = json.dumps({
+            "rows": [
+                {"row_id": "0", "detected_source_language": "fr", "ru": "текст", "en": "text"},
+                {"row_id": "1", "detected_source_language": "en", "ru": "дашборд", "en": "dashboard"},
+            ]
+        })
+        batch_rows = [
+            {"row_index": "0", "source_text": "texte", "source_object_name": "Row 0"},
+            {"row_index": "1", "source_text": "dashboard", "source_object_name": "Row 1"},
+        ]
+
+        with patch.object(executor, '_call_llm', return_value=multi_lang_response):
+            result = executor._call_llm_for_batch(
+                job=job,
+                run_id="run-ml-1",
+                batch_rows=batch_rows,
+                dict_matches=[],
+                batch_id="batch-ml-1",
+            )
+
+        assert result["successful"] == 2
+        assert result["failed"] == 0
+        assert result["skipped"] == 0
+
+        # Check that TranslationLanguage entries were created for each row per language
+        # Each row should have 2 language entries (ru, en) = 4 total
+        lang_added_calls = [
+            call for call in db.add.call_args_list
+            if isinstance(call[0][0], TranslationLanguage)
+        ]
+        # Should have 4 language entries (2 rows x 2 languages)
+        assert len(lang_added_calls) >= 4, f"Expected at least 4 TranslationLanguage entries, got {len(lang_added_calls)}"
+
+        # Group by language code
+        lang_codes: list[str] = []
+        for call in lang_added_calls:
+            lang_entry = call[0][0]
+            lang_codes.append(lang_entry.language_code)
+        assert "ru" in lang_codes
+        assert "en" in lang_codes
+        # Each language should appear twice (once per row)
+        assert lang_codes.count("ru") == 2
+        assert lang_codes.count("en") == 2
+    # endregion test_multi_language_translation_language_entries
+
+    # region test_source_as_reference [TYPE Function]
+    # @PURPOSE: When detected source language matches a target language, store original text verbatim.
+    def test_source_as_reference(self) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        job = MagicMock(spec=TranslationJob)
+        job.id = "job-sar-1"
+        job.target_languages = ["fr", "en"]
+        job.target_language = None
+        job.target_dialect = "postgresql"
+        job.source_dialect = "postgresql"
+        job.translation_column = "name"
+        job.provider_id = "provider-1"
+        job.batch_size = 50
+
+        executor = TranslationExecutor(db, config_manager, "test-user")
+
+        # Source is French, target_languages include French
+        response = json.dumps({
+            "rows": [
+                {"row_id": "0", "detected_source_language": "fr", "fr": "texte français", "en": "french text"},
+            ]
+        })
+        batch_rows = [
+            {"row_index": "0", "source_text": "texte français", "source_object_name": "Row 0"},
+        ]
+
+        with patch.object(executor, '_call_llm', return_value=response):
+            result = executor._call_llm_for_batch(
+                job=job,
+                run_id="run-sar-1",
+                batch_rows=batch_rows,
+                dict_matches=[],
+                batch_id="batch-sar-1",
+            )
+
+        assert result["successful"] == 1
+
+        # Find the TranslationLanguage for French — should use source text (original)
+        lang_calls = [
+            call for call in db.add.call_args_list
+            if isinstance(call[0][0], TranslationLanguage)
+        ]
+        fr_entry = None
+        en_entry = None
+        for call in lang_calls:
+            le = call[0][0]
+            if le.language_code == "fr":
+                fr_entry = le
+            elif le.language_code == "en":
+                en_entry = le
+
+        assert fr_entry is not None, "FR language entry should exist"
+        # Source-as-reference: fr translation should be the original source text
+        assert fr_entry.translated_value == "texte français", (
+            f"Expected source text 'texte français' for fr, got '{fr_entry.translated_value}'"
+        )
+        assert en_entry is not None, "EN language entry should exist"
+        # EN should get the LLM-translated value
+        assert en_entry.translated_value == "french text"
+    # endregion test_source_as_reference
+
+    # region test_per_language_stats_on_execute_run [TYPE Function]
+    # @PURPOSE: execute_run with multi-language job creates and populates TranslationRunLanguageStats.
+    def test_per_language_stats_on_execute_run(self, mock_job: MagicMock) -> None:
+        db = MagicMock()
+        config_manager = MagicMock()
+
+        # Configure mock_job for multi-language
+        mock_job.target_languages = ["ru", "en"]
+        mock_job.target_language = None
+        mock_job.target_table = "target_tbl"
+
+        # Mock run
+        run = MagicMock()
+        run.id = "run-stats-1"
+        run.job_id = "job-123"
+        run.status = "PENDING"
+
+        # Mock completed run returned by executor
+        completed_run = MagicMock()
+        completed_run.id = "run-stats-1"
+        completed_run.job_id = "job-123"
+        completed_run.status = "COMPLETED"
+        completed_run.total_records = 5
+        completed_run.successful_records = 5
+        completed_run.failed_records = 0
+        completed_run.skipped_records = 0
+        completed_run.completed_at = datetime.now(UTC)
+        completed_run.error_message = None
+        completed_run.insert_status = "success"
+        completed_run.superset_execution_id = "q-1"
+        completed_run.superset_execution_log = {}
+
+        # Mock queries: job -> mock_job, run -> completed_run
+        db.query.return_value.filter.return_value.first.side_effect = [
+            mock_job,
+            completed_run,
+        ]
+
+        orch = TranslationOrchestrator(db, config_manager, "test-user")
+        with patch.object(orch, 'event_log'), \
+             patch.object(orch, '_generate_and_insert_sql',
+                          return_value={"status": "success", "query_id": "q-1", "rows_affected": 5}), \
+             patch.object(orch, '_update_language_stats') as mock_update_stats, \
+             patch('src.plugins.translate.orchestrator.TranslationExecutor') as MockExecutor:
+
+            mock_executor_instance = MagicMock()
+            mock_executor_instance.execute_run.return_value = completed_run
+            MockExecutor.return_value = mock_executor_instance
+
+            result = orch.execute_run(run)
+
+        assert result.status == "COMPLETED"
+
+        # Verify that TranslationRunLanguageStats entries were created
+        stats_added = [
+            call for call in db.add.call_args_list
+            if isinstance(call[0][0], TranslationRunLanguageStats)
+        ]
+        # 2 language stats entries (ru, en)
+        assert len(stats_added) >= 2, (
+            f"Expected at least 2 TranslationRunLanguageStats entries, got {len(stats_added)}"
+        )
+        codes = [call[0][0].language_code for call in stats_added]
+        assert "ru" in codes
+        assert "en" in codes
+
+        # Verify _update_language_stats was called
+        mock_update_stats.assert_called_once()
+    # endregion test_per_language_stats_on_execute_run
+
+# endregion TestTranslationExecutorMultiLang
 # endregion OrchestratorTests

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

@@ -420,8 +420,13 @@ class TestTranslationPreview:
         assert tokens > 0
         assert tokens < len(prompt)  # tokens typically fewer than chars
 
-        output_tokens = TokenEstimator.estimate_output_tokens(10)
-        assert output_tokens == 500
+        # Single language
+        output_tokens = TokenEstimator.estimate_output_tokens(10, num_languages=1)
+        assert output_tokens == 600  # 10 * 1 * 50 * 1.2
+
+        # Multi-language
+        output_tokens_multi = TokenEstimator.estimate_output_tokens(10, num_languages=3)
+        assert output_tokens_multi == 1800  # 10 * 3 * 50 * 1.2
 
         cost = TokenEstimator.estimate_cost(1000, 0.002)
         assert cost == 0.002
@@ -431,28 +436,65 @@ class TestTranslationPreview:
     # endregion test_cost_estimation
 
     # region test_preview_parse_llm_response [TYPE Function]
-    # @PURPOSE: Verify LLM JSON response parsing.
+    # @PURPOSE: Verify LLM JSON response parsing includes detected_source_language.
     def test_preview_parse_llm_response(self):
-        """Parse LLM response should extract translations keyed by row_id."""
+        """Parse LLM response should extract translations with detected_source_language per row."""
         response = json.dumps({
             "rows": [
-                {"row_id": "0", "translation": "Привет"},
+                {"row_id": "0", "translation": "Привет", "detected_source_language": "en"},
                 {"row_id": "1", "translation": "Мир"},
             ]
         })
         result = TranslationPreview._parse_llm_response(response, 2)
-        assert result == {"0": "Привет", "1": "Мир"}
+        assert result == {
+            "0": {"translation": "Привет", "detected_source_language": "en"},
+            "1": {"translation": "Мир", "detected_source_language": "und"},
+        }
     # endregion test_preview_parse_llm_response
 
+    # region test_preview_parse_llm_response_multilang [TYPE Function]
+    # @PURPOSE: Verify multi-language LLM response parsing.
+    def test_preview_parse_llm_response_multilang(self):
+        """Parse multi-language LLM response should extract per-language translations."""
+        response = json.dumps({
+            "rows": [
+                {"row_id": "0", "detected_source_language": "fr", "ru": "текст", "en": "text", "de": "Text"},
+                {"row_id": "1", "detected_source_language": "en", "ru": "привет", "en": "hello"},
+            ]
+        })
+        result = TranslationPreview._parse_llm_response(response, 2, target_languages=["ru", "en", "de"])
+        assert result == {
+            "0": {"detected_source_language": "fr", "ru": "текст", "en": "text", "de": "Text"},
+            "1": {"detected_source_language": "en", "ru": "привет", "en": "hello"},
+        }
+        # Verify per-language extraction
+        assert result["0"]["ru"] == "текст"
+        assert result["0"]["en"] == "text"
+        assert result["1"]["ru"] == "привет"
+    # endregion test_preview_parse_llm_response_multilang
+
     # region test_preview_parse_llm_response_with_code_block [TYPE Function]
     # @PURPOSE: Verify LLM response parsing handles markdown code blocks.
     def test_preview_parse_llm_response_with_code_block(self):
         """Parse LLM response should handle markdown code block wrapping."""
         response = "```json\n{\n    \"rows\": [\n        {\"row_id\": \"0\", \"translation\": \"Test\"}\n    ]\n}\n```"
         result = TranslationPreview._parse_llm_response(response, 1)
-        assert result == {"0": "Test"}
+        assert result == {"0": {"translation": "Test", "detected_source_language": "und"}}
     # endregion test_preview_parse_llm_response_with_code_block
 
+    # region test_cost_warning [TYPE Function]
+    # @PURPOSE: Verify cost warning for large sample sizes.
+    def test_cost_warning(self):
+        """Cost warning should appear for sample_size > 30."""
+        warning = TokenEstimator.check_cost_warning(31, 2, 5000, 0.01)
+        assert warning is not None
+        assert "Large preview" in warning
+        assert "5000" in warning
+
+        no_warning = TokenEstimator.check_cost_warning(10, 2, 1000, 0.002)
+        assert no_warning is None
+    # endregion test_cost_warning
+
     # region test_preview_compute_config_hash [TYPE Function]
     # @PURPOSE: Verify config hash computation is deterministic.
     def test_preview_compute_config_hash(self):

backend/src/plugins/translate/dictionary.py

@@ -14,7 +14,7 @@ import io
 import re
 from typing import Any
 
-from sqlalchemy import func
+from sqlalchemy import func, or_
 from sqlalchemy.orm import Session
 
 from ...core.logger import belief_scope, logger
@@ -27,6 +27,23 @@ from ...models.translate import (
 from ._utils import _detect_delimiter, _normalize_term
 
 
+# #region _validate_bcp47 [C:2] [TYPE Function] [SEMANTICS validation, bcp47, language]
+# @BRIEF Validate that a language tag is a non-empty BCP-47 string.
+def _validate_bcp47(tag: str, field_name: str) -> None:
+    """Validate that tag is a non-empty BCP-47 string (basic check)."""
+    if not tag or not tag.strip():
+        raise ValueError(f"{field_name} must be a non-empty BCP-47 language tag")
+    tag = tag.strip()
+    # Basic BCP-47 validation: must match lang[-script][-region][-variant]* pattern
+    import re as _re
+    if not _re.match(r'^[a-zA-Z]{2,8}(-[a-zA-Z0-9]{1,8})*$', tag):
+        raise ValueError(
+            f"{field_name} is not a valid BCP-47 tag: '{tag}'. "
+            "Expected format like 'en', 'ru', 'zh-CN', 'pt-BR'."
+        )
+# #endregion _validate_bcp47
+
+
 # #region DictionaryManager [C:4] [TYPE Class]
 # @BRIEF Manages terminology dictionaries and their entries with referential integrity.
 # @PRE: Database session is open and valid.
@@ -34,12 +51,14 @@ from ._utils import _detect_delimiter, _normalize_term
 # @SIDE_EFFECT: Creates, updates, deletes TerminologyDictionary and DictionaryEntry rows; enforces deletion guards.
 class DictionaryManager:
     # region DictionaryManager.create_dictionary [TYPE Function]
-    # @PURPOSE: Create a new terminology dictionary.
-    # @PRE: payload contains name, source_dialect, target_dialect.
+    # @PURPOSE: Create a new terminology dictionary (language independent at dictionary level; language pairs are per-entry).
+    # @PRE: name is provided.
     # @POST: New TerminologyDictionary row is created and returned.
     @staticmethod
     def create_dictionary(
-        db: Session, name: str, source_dialect: str, target_dialect: str,
+        db: Session, name: str,
+        source_dialect: str = "",
+        target_dialect: str = "",
         created_by: str | None = None, description: str | None = None,
         is_active: bool = True,
     ) -> TerminologyDictionary:
@@ -48,8 +67,8 @@ class DictionaryManager:
             dictionary = TerminologyDictionary(
                 name=name,
                 description=description,
-                source_dialect=source_dialect,
-                target_dialect=target_dialect,
+                source_dialect=source_dialect or "",
+                target_dialect=target_dialect or "",
                 is_active=is_active,
                 created_by=created_by,
             )
@@ -166,53 +185,66 @@ class DictionaryManager:
     # endregion DictionaryManager.list_dictionaries
 
     # region DictionaryManager.add_entry [TYPE Function]
-    # @PURPOSE: Add an entry to a dictionary, enforcing unique source_term_normalized.
-    # @PRE: dict_id exists. source_term and target_term are non-empty.
+    # @PURPOSE: Add an entry to a dictionary with language-pair-aware uniqueness validation.
+    # @PRE: dict_id exists. source_term, target_term are non-empty. source_language, target_language are valid BCP-47.
     # @POST: New DictionaryEntry row is created or raises on duplicate.
     @staticmethod
     def add_entry(
         db: Session, dict_id: str, source_term: str, target_term: str,
         context_notes: str | None = None,
+        source_language: str = "und",
+        target_language: str = "und",
     ) -> DictionaryEntry:
         with belief_scope("DictionaryManager.add_entry"):
+            _validate_bcp47(source_language, "source_language")
+            _validate_bcp47(target_language, "target_language")
+
             normalized = _normalize_term(source_term)
+            # Uniqueness is now (dictionary_id, source_term_normalized, source_language, target_language)
             existing = (
                 db.query(DictionaryEntry)
                 .filter(
                     DictionaryEntry.dictionary_id == dict_id,
                     DictionaryEntry.source_term_normalized == normalized,
+                    DictionaryEntry.source_language == source_language,
+                    DictionaryEntry.target_language == target_language,
                 )
                 .first()
             )
             if existing:
                 raise ValueError(
-                    f"Duplicate entry: '{source_term}' already exists in this dictionary "
-                    f"(id={existing.id}). Use overwrite or keep_existing conflict mode."
+                    f"Duplicate entry: '{source_term}' (lang: {source_language}→{target_language}) "
+                    f"already exists in this dictionary (id={existing.id}). "
+                    "Use overwrite or keep_existing conflict mode."
                 )
 
-            logger.reason("Adding dictionary entry", {"dict_id": dict_id, "term": source_term})
+            logger.reason("Adding dictionary entry", {"dict_id": dict_id, "term": source_term, "src_lang": source_language, "tgt_lang": target_language})
             entry = DictionaryEntry(
                 dictionary_id=dict_id,
                 source_term=source_term.strip(),
                 source_term_normalized=normalized,
                 target_term=target_term.strip(),
                 context_notes=context_notes.strip() if context_notes else None,
+                source_language=source_language.strip(),
+                target_language=target_language.strip(),
             )
             db.add(entry)
             db.commit()
             db.refresh(entry)
-            logger.reflect("Entry added", {"entry_id": entry.id})
+            logger.reflect("Entry added", {"entry_id": entry.id, "src_lang": source_language, "tgt_lang": target_language})
             return entry
     # endregion DictionaryManager.add_entry
 
     # region DictionaryManager.edit_entry [TYPE Function]
-    # @PURPOSE: Edit an existing dictionary entry with duplicate-aware normalization.
+    # @PURPOSE: Edit an existing dictionary entry with language-pair-aware duplicate check.
     # @PRE: entry_id exists.
     # @POST: Entry fields are updated.
     @staticmethod
     def edit_entry(
         db: Session, entry_id: str, source_term: str | None = None,
         target_term: str | None = None, context_notes: str | None = None,
+        source_language: str | None = None,
+        target_language: str | None = None,
     ) -> DictionaryEntry:
         with belief_scope("DictionaryManager.edit_entry"):
             entry = db.query(DictionaryEntry).filter(DictionaryEntry.id == entry_id).first()
@@ -220,14 +252,24 @@ class DictionaryManager:
                 raise ValueError(f"Entry not found: {entry_id}")
 
             logger.reason("Editing dictionary entry", {"entry_id": entry_id})
+
+            if source_language is not None:
+                _validate_bcp47(source_language, "source_language")
+                entry.source_language = source_language.strip()
+            if target_language is not None:
+                _validate_bcp47(target_language, "target_language")
+                entry.target_language = target_language.strip()
+
             if source_term is not None:
                 normalized = _normalize_term(source_term)
-                # Check uniqueness within the same dictionary
+                # Check uniqueness within the same dictionary + language pair
                 existing = (
                     db.query(DictionaryEntry)
                     .filter(
                         DictionaryEntry.dictionary_id == entry.dictionary_id,
                         DictionaryEntry.source_term_normalized == normalized,
+                        DictionaryEntry.source_language == entry.source_language,
+                        DictionaryEntry.target_language == entry.target_language,
                         DictionaryEntry.id != entry_id,
                     )
                     .first()
@@ -318,6 +360,8 @@ class DictionaryManager:
         delimiter: str | None = None,
         on_conflict: str = "overwrite",
         preview_only: bool = False,
+        default_source_language: str | None = None,
+        default_target_language: str | None = None,
     ) -> dict[str, Any]:
         with belief_scope("DictionaryManager.import_entries"):
             # Validate dictionary
@@ -359,6 +403,8 @@ class DictionaryManager:
                     source_term = row.get("source_term", "").strip()
                     target_term = row.get("target_term", "").strip()
                     context_notes = row.get("context_notes", "").strip() or None
+                    source_language = row.get("source_language", "").strip() or default_source_language or "und"
+                    target_language = row.get("target_language", "").strip() or default_target_language or "und"
 
                     if not source_term or not target_term:
                         result["errors"].append({
@@ -376,6 +422,8 @@ class DictionaryManager:
                             .filter(
                                 DictionaryEntry.dictionary_id == dict_id,
                                 DictionaryEntry.source_term_normalized == normalized,
+                                DictionaryEntry.source_language == source_language,
+                                DictionaryEntry.target_language == target_language,
                             )
                             .first()
                         )
@@ -384,6 +432,8 @@ class DictionaryManager:
                             "source_term": source_term,
                             "target_term": target_term,
                             "context_notes": context_notes,
+                            "source_language": source_language,
+                            "target_language": target_language,
                             "is_conflict": existing is not None,
                             "existing_target_term": existing.target_term if existing else None,
                         }
@@ -395,6 +445,8 @@ class DictionaryManager:
                         .filter(
                             DictionaryEntry.dictionary_id == dict_id,
                             DictionaryEntry.source_term_normalized == normalized,
+                            DictionaryEntry.source_language == source_language,
+                            DictionaryEntry.target_language == target_language,
                         )
                         .first()
                     )
@@ -404,6 +456,8 @@ class DictionaryManager:
                             existing.source_term = source_term
                             existing.target_term = target_term
                             existing.context_notes = context_notes
+                            existing.source_language = source_language
+                            existing.target_language = target_language
                             result["updated"] += 1
                         elif on_conflict == "keep_existing":
                             result["skipped"] += 1
@@ -421,6 +475,8 @@ class DictionaryManager:
                             source_term_normalized=normalized,
                             target_term=target_term,
                             context_notes=context_notes,
+                            source_language=source_language,
+                            target_language=target_language,
                         )
                         db.add(entry)
                         result["created"] += 1
@@ -446,14 +502,122 @@ class DictionaryManager:
             return result
     # endregion DictionaryManager.import_entries
 
+    # region DictionaryManager.export_entries [TYPE Function]
+    # @PURPOSE: Export all entries as CSV string with language columns.
+    # @PRE: dict_id exists.
+    # @POST: Returns CSV string with header: source_term, target_term, source_language, target_language, context_notes, context_data, usage_notes.
+    @staticmethod
+    def export_entries(
+        db: Session, dict_id: str, delimiter: str = ",",
+    ) -> str:
+        with belief_scope("DictionaryManager.export_entries"):
+            dictionary = db.query(TerminologyDictionary).filter(TerminologyDictionary.id == dict_id).first()
+            if not dictionary:
+                raise ValueError(f"Dictionary not found: {dict_id}")
+
+            entries = (
+                db.query(DictionaryEntry)
+                .filter(DictionaryEntry.dictionary_id == dict_id)
+                .order_by(DictionaryEntry.source_term.asc())
+                .all()
+            )
+
+            output = io.StringIO()
+            fieldnames = [
+                "source_term", "target_term",
+                "source_language", "target_language",
+                "context_notes", "context_data", "usage_notes",
+            ]
+            writer = csv.DictWriter(output, fieldnames=fieldnames, delimiter=delimiter)
+            writer.writeheader()
+
+            for entry in entries:
+                writer.writerow({
+                    "source_term": entry.source_term,
+                    "target_term": entry.target_term,
+                    "source_language": entry.source_language,
+                    "target_language": entry.target_language,
+                    "context_notes": entry.context_notes or "",
+                    "context_data": entry.context_data,
+                    "usage_notes": entry.usage_notes or "",
+                })
+
+            result = output.getvalue()
+            logger.reflect("Entries exported", {"dict_id": dict_id, "count": len(entries), "delimiter": repr(delimiter)})
+            return result
+    # endregion DictionaryManager.export_entries
+
+    # region DictionaryManager.migrate_old_entries [TYPE Function]
+    # @PURPOSE: Migrate existing single-language dictionary entries to populate source_language and target_language.
+    # @PRE: db session is open.
+    # @POST: Entries with "und" source_language or target_language are updated with inferred values.
+    # @SIDE_EFFECT: Updates DictionaryEntry rows in bulk.
+    @staticmethod
+    def migrate_old_entries(db: Session) -> dict[str, int]:
+        with belief_scope("DictionaryManager.migrate_old_entries"):
+            logger.reason("Starting migration of old dictionary entries")
+            migrated_source = 0
+            migrated_target = 0
+            skipped = 0
+
+            # Find all entries that may need migration (source_language is "und" or target_language is "und")
+            entries = (
+                db.query(DictionaryEntry)
+                .filter(
+                    (DictionaryEntry.source_language == "und") |
+                    (DictionaryEntry.target_language == "und")
+                )
+                .all()
+            )
+
+            for entry in entries:
+                # Try to infer source_language from origin_source_language
+                if entry.source_language == "und":
+                    if entry.origin_source_language:
+                        entry.source_language = entry.origin_source_language
+                        migrated_source += 1
+
+                # Try to infer target_language from the parent dictionary's deprecated target_language
+                if entry.target_language == "und":
+                    dictionary = (
+                        db.query(TerminologyDictionary)
+                        .filter(TerminologyDictionary.id == entry.dictionary_id)
+                        .first()
+                    )
+                    if dictionary and dictionary.target_language:
+                        entry.target_language = dictionary.target_language
+                        migrated_target += 1
+
+                if entry.source_language == "und" and entry.target_language == "und":
+                    skipped += 1
+
+            db.commit()
+
+            logger.reflect("Migration complete", {
+                "migrated_source": migrated_source,
+                "migrated_target": migrated_target,
+                "skipped": skipped,
+                "total_processed": len(entries),
+            })
+            return {
+                "migrated_source": migrated_source,
+                "migrated_target": migrated_target,
+                "skipped": skipped,
+                "total_processed": len(entries),
+            }
+    # endregion DictionaryManager.migrate_old_entries
+
     # region DictionaryManager.filter_for_batch [TYPE Function]
-    # @PURPOSE: Scan batch texts for case-insensitive, word-boundary-aware matches against all dictionaries attached to a job.
+    # @PURPOSE: Scan batch texts for case-insensitive, word-boundary-aware matches against all dictionaries attached to a job,
+    #           optionally filtered by language pair.
     # @PRE: job_id exists and source_texts is a list of strings.
     # @POST: Returns list of matched entries with match info, sorted by priority across attached dictionaries.
     # @SIDE_EFFECT: Queries TranslationJobDictionary, TerminologyDictionary, and DictionaryEntry tables.
     @staticmethod
     def filter_for_batch(
         db: Session, source_texts: list[str], job_id: str,
+        source_language: str | None = None,
+        target_language: str | None = None,
     ) -> list[dict[str, Any]]:
         with belief_scope("DictionaryManager.filter_for_batch"):
             # Get dictionaries attached to this job
@@ -508,6 +672,18 @@ class DictionaryManager:
                     if not norm:
                         continue
 
+                    # Apply language pair filtering
+                    if source_language is not None:
+                        src = source_language.strip().lower()
+                        entry_src = entry.source_language.strip().lower()
+                        if entry_src != src and entry_src != "und":
+                            continue
+                    if target_language is not None:
+                        tgt = target_language.strip().lower()
+                        entry_tgt = entry.target_language.strip().lower()
+                        if entry_tgt != tgt:
+                            continue
+
                     # Word-boundary-aware matching
                     # Build pattern: \bterm\b (case-insensitive)
                     # Escape regex special chars in the search term
@@ -528,6 +704,12 @@ class DictionaryManager:
                                 "dictionary_id": entry.dictionary_id,
                                 "dictionary_name": dict_map.get(entry.dictionary_id, ""),
                                 "context_notes": entry.context_notes,
+                                "context_data": entry.context_data,
+                                "has_context": entry.has_context,
+                                "context_source": entry.context_source,
+                                "usage_notes": entry.usage_notes,
+                                "source_language": entry.source_language,
+                                "target_language": entry.target_language,
                             })
 
             # Sort by dictionary link priority (order of dict_ids from link order)
@@ -539,6 +721,8 @@ class DictionaryManager:
                 "source_texts": len(source_texts),
                 "matches": len(matched),
                 "dictionaries_used": len(dictionaries),
+                "source_language": source_language,
+                "target_language": target_language,
             })
             return matched
     # endregion DictionaryManager.filter_for_batch
@@ -560,6 +744,8 @@ class DictionaryManager:
         origin_row_key: str | None = None,
         origin_user_id: str | None = None,
         on_conflict: str = "overwrite",
+        context_data: dict[str, Any] | None = None,
+        usage_notes: str | None = None,
     ) -> dict[str, Any]:
         with belief_scope("DictionaryManager.submit_correction"):
             # Validate dictionary exists
@@ -568,11 +754,22 @@ class DictionaryManager:
                 raise ValueError(f"Dictionary '{dict_id}' not found")
 
             normalized = _normalize_term(source_term)
+            entry_src_lang = dictionary.source_dialect or "und"
+            entry_tgt_lang = dictionary.target_dialect or "und"
+            # Find existing entry: match exact language pair OR old-style "und"/"und" entries
             existing = (
                 db.query(DictionaryEntry)
                 .filter(
                     DictionaryEntry.dictionary_id == dict_id,
                     DictionaryEntry.source_term_normalized == normalized,
+                    or_(
+                        # Exact language pair match
+                        (DictionaryEntry.source_language == entry_src_lang) &
+                        (DictionaryEntry.target_language == entry_tgt_lang),
+                        # Old-style entries without language pair
+                        (DictionaryEntry.source_language == "und") &
+                        (DictionaryEntry.target_language == "und"),
+                    ),
                 )
                 .first()
             )
@@ -606,6 +803,12 @@ class DictionaryManager:
                         existing.origin_row_key = origin_row_key
                     if origin_user_id:
                         existing.origin_user_id = origin_user_id
+                    if context_data is not None:
+                        existing.context_data = context_data
+                        existing.has_context = bool(context_data)
+                        existing.context_source = "auto"
+                    if usage_notes is not None:
+                        existing.usage_notes = usage_notes
                     db.flush()
                     result["action"] = "updated"
                     result["entry_id"] = existing.id
@@ -621,12 +824,18 @@ class DictionaryManager:
                     result["message"] = "Correction cancelled by conflict mode"
                     return result
             else:
-                # Create new entry
+                # Create new entry — derive language pair from dictionary
                 entry = DictionaryEntry(
                     dictionary_id=dict_id,
                     source_term=source_term.strip(),
                     source_term_normalized=normalized,
                     target_term=corrected_target_term.strip(),
+                    source_language=entry_src_lang,
+                    target_language=entry_tgt_lang,
+                    context_data=context_data,
+                    usage_notes=usage_notes,
+                    has_context=bool(context_data),
+                    context_source="auto" if context_data else None,
                     origin_run_id=origin_run_id,
                     origin_row_key=origin_row_key,
                     origin_user_id=origin_user_id,
@@ -678,11 +887,19 @@ class DictionaryManager:
                     continue
 
                 normalized = _normalize_term(source_term)
+                bulk_src_lang = dictionary.source_dialect or "und"
+                bulk_tgt_lang = dictionary.target_dialect or "und"
                 existing = (
                     db.query(DictionaryEntry)
                     .filter(
                         DictionaryEntry.dictionary_id == dict_id,
                         DictionaryEntry.source_term_normalized == normalized,
+                        or_(
+                            (DictionaryEntry.source_language == bulk_src_lang) &
+                            (DictionaryEntry.target_language == bulk_tgt_lang),
+                            (DictionaryEntry.source_language == "und") &
+                            (DictionaryEntry.target_language == "und"),
+                        ),
                     )
                     .first()
                 )
@@ -729,6 +946,8 @@ class DictionaryManager:
                         source_term=source_term,
                         source_term_normalized=normalized,
                         target_term=corrected_target,
+                        source_language=bulk_src_lang,
+                        target_language=bulk_tgt_lang,
                         origin_run_id=corr.get("origin_run_id"),
                         origin_row_key=corr.get("origin_row_key"),
                         origin_user_id=origin_user_id,

backend/src/plugins/translate/events.py

@@ -191,6 +191,18 @@ class TranslationEventLog:
                 logger.reflect("No expired events to prune", {})
                 return {"pruned": 0, "snapshot_id": None}
 
+            # Compute per-language metrics from expired events before pruning
+            per_language: dict[str, dict[str, int | float]] = {}
+            expired_events = expired_query.all()
+            for evt in expired_events:
+                event_data = evt.event_data or {}
+                lang = event_data.get("language_code") or event_data.get("target_language") or "_unknown_"
+                if lang not in per_language:
+                    per_language[lang] = {"cumulative_tokens": 0, "cumulative_cost": 0.0, "runs": 0}
+                per_language[lang]["cumulative_tokens"] += event_data.get("token_count", 0) or 0
+                per_language[lang]["cumulative_cost"] += event_data.get("cost", 0.0) or 0.0
+                per_language[lang]["runs"] += 1 if event_data.get("run_id") else 0
+
             # Create MetricSnapshot before pruning
             snapshot = MetricSnapshot(
                 id=str(uuid.uuid4()),
@@ -198,6 +210,7 @@ class TranslationEventLog:
                 key_hash=f"prune_{cutoff.timestamp():.0f}",
                 covers_events_before=cutoff,
                 total_records=total_expired,
+                per_language_metrics=per_language or None,
                 snapshot_date=datetime.now(UTC),
             )
             self.db.add(snapshot)

backend/src/plugins/translate/executor.py

@@ -27,6 +27,7 @@ from ...core.logger import belief_scope, logger
 from ...models.translate import (
     TranslationBatch,
     TranslationJob,
+    TranslationLanguage,
     TranslationPreviewRecord,
     TranslationPreviewSession,
     TranslationRecord,
@@ -36,6 +37,7 @@ from ...services.llm_prompt_templates import render_prompt
 from ...services.llm_provider import LLMProviderService
 from .dictionary import DictionaryManager
 from .preview import DEFAULT_EXECUTION_PROMPT_TEMPLATE
+from .prompt_builder import ContextAwarePromptBuilder
 
 # #region MAX_RETRIES_PER_BATCH [TYPE Constant]
 # @BRIEF Maximum number of retries for a single batch before marking it failed.
@@ -62,6 +64,7 @@ class TranslationExecutor:
         self.current_user = current_user
         self.on_batch_progress = on_batch_progress
         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 execute_run [TYPE Function]
     # @PURPOSE: Run full translation execution for a TranslationRun.
@@ -84,6 +87,9 @@ class TranslationExecutor:
                 "batch_size": job.batch_size,
             })
 
+            # Load preview edits for carry-forward
+            self._load_preview_edits(job.id)
+
             # Mark run as RUNNING
             run.status = "RUNNING"
             run.started_at = datetime.now(UTC)
@@ -410,6 +416,70 @@ class TranslationExecutor:
         return []
     # endregion _extract_chart_data_rows
 
+    # region _load_preview_edits [TYPE Function]
+    # @PURPOSE: Load user edits from accepted preview session for carry-forward.
+    # @PRE: job_id exists.
+    # @POST: Populates _preview_edits_cache with key_hash -> {lang_code: edited_value}.
+    # @SIDE_EFFECT: Queries TranslationPreviewLanguage and TranslationPreviewRecord.
+    def _load_preview_edits(self, job_id: str) -> None:
+        """Load preview edits for carry-forward during execution."""
+        from ...models.translate import TranslationPreviewLanguage, TranslationPreviewRecord, TranslationPreviewSession
+
+        with belief_scope("TranslationExecutor._load_preview_edits"):
+            session = (
+                self.db.query(TranslationPreviewSession)
+                .filter(
+                    TranslationPreviewSession.job_id == job_id,
+                    TranslationPreviewSession.status == "APPLIED",
+                )
+                .order_by(TranslationPreviewSession.created_at.desc())
+                .first()
+            )
+            if not session:
+                logger.reason("No applied preview session found — no edits to carry forward", {
+                    "job_id": job_id,
+                })
+                self._preview_edits_cache = {}
+                return
+
+            records = (
+                self.db.query(TranslationPreviewRecord)
+                .filter(TranslationPreviewRecord.session_id == session.id)
+                .all()
+            )
+
+            edits: dict[str, dict[str, str]] = {}
+            for rec in records:
+                if not rec.source_data:
+                    continue
+                # Compute key hash from source_data
+                key_hash = self._compute_key_hash(rec.source_data)
+                edited_langs: dict[str, str] = {}
+                for lang_entry in (rec.languages or []):
+                    if lang_entry.status in ("edited", "approved") and lang_entry.user_edit:
+                        edited_langs[lang_entry.language_code] = lang_entry.final_value or lang_entry.user_edit
+                        logger.reason("Carrying forward preview edit", {
+                            "key_hash": key_hash,
+                            "language_code": lang_entry.language_code,
+                        })
+                if edited_langs:
+                    edits[key_hash] = edited_langs
+
+            self._preview_edits_cache = edits
+            logger.reason(f"Loaded {len(edits)} preview edits for carry-forward", {
+                "job_id": job_id,
+            })
+    # endregion _load_preview_edits
+
+    # region _compute_key_hash [TYPE Function]
+    # @PURPOSE: Compute a stable hash from source_data dict for matching preview edits.
+    @staticmethod
+    def _compute_key_hash(source_data: dict) -> str:
+        import hashlib
+        stable = json.dumps(source_data, sort_keys=True)
+        return hashlib.sha256(stable.encode()).hexdigest()[:16]
+    # endregion _compute_key_hash
+
     # region _process_batch [TYPE Function]
     # @PURPOSE: Process a single batch: filter dict, build prompt, call LLM, persist records.
     # @PRE: job and batch_rows are valid.
@@ -452,17 +522,39 @@ class TranslationExecutor:
                 self.db, source_texts, job.id
             )
 
-            # For each row, determine if we need LLM translation or can use approved translation
+            # For each row, determine if we need LLM translation or can use approved/preview edit
             rows_for_llm = []
             pre_translated = []
 
             for row in batch_rows:
                 if row.get("approved_translation"):
                     pre_translated.append(row)
-                else:
-                    rows_for_llm.append(row)
+                    continue
+
+                # Check for preview edits carry-forward
+                source_data = row.get("source_data") or {}
+                if source_data and self._preview_edits_cache:
+                    key_hash = self._compute_key_hash(source_data)
+                    preview_edit = self._preview_edits_cache.get(key_hash)
+                    if preview_edit:
+                        # Use the first edited language's value as the approved translation
+                        first_edit = next(iter(preview_edit.values()), None)
+                        if first_edit:
+                            row["approved_translation"] = first_edit
+                            logger.reason("Using preview edit carry-forward", {
+                                "key_hash": key_hash,
+                                "langs": list(preview_edit.keys()),
+                            })
+                            pre_translated.append(row)
+                            continue
+
+                rows_for_llm.append(row)
 
             # Handle pre-translated (approved) rows
+            target_languages = job.target_languages or [job.target_language or job.target_dialect or "en"]
+            if not isinstance(target_languages, list):
+                target_languages = [str(target_languages)]
+
             for row in pre_translated:
                 record = TranslationRecord(
                     id=str(uuid.uuid4()),
@@ -477,6 +569,20 @@ class TranslationExecutor:
                     status="SUCCESS",
                 )
                 self.db.add(record)
+
+                # Create per-language entry for each target language (pre-approved)
+                for lang_code in target_languages:
+                    lang_entry = TranslationLanguage(
+                        id=str(uuid.uuid4()),
+                        record_id=record.id,
+                        language_code=lang_code,
+                        source_language_detected="und",
+                        translated_value=row.get("approved_translation"),
+                        final_value=row.get("approved_translation"),
+                        status="translated",
+                        needs_review=False,
+                    )
+                    self.db.add(lang_entry)
                 result["successful"] += 1
 
             # Process rows needing LLM translation
@@ -524,21 +630,31 @@ class TranslationExecutor:
         batch_id: str,
     ) -> dict[str, int]:
         with belief_scope("TranslationExecutor._call_llm_for_batch"):
-            # Build dictionary section
+            # Build dictionary section using ContextAwarePromptBuilder
             dictionary_section = ""
             if dict_matches:
+                # Get row_context from first batch row if available
+                row_context = batch_rows[0].get("source_data") if batch_rows else None
+
+                # Use ContextAwarePromptBuilder for context-aware annotations
+                annotated_entries = ContextAwarePromptBuilder.build_context_entries(
+                    dict_matches, row_context
+                )
                 glossary_lines = []
-                for m in dict_matches:
-                    glossary_lines.append(
-                        f"- '{m['source_term']}' -> '{m['target_term']}'"
-                        f"{' (' + m['context_notes'] + ')' if m.get('context_notes') else ''}"
-                    )
+                for annotated_line in annotated_entries:
+                    glossary_lines.append(f"- {annotated_line}")
                 dictionary_section = (
                     "Terminology dictionary (use these translations when applicable):\n"
                     + "\n".join(glossary_lines)
                     + "\n\n"
                 )
 
+            # Resolve target languages
+            target_languages = job.target_languages or [job.target_language or job.target_dialect or "en"]
+            if not isinstance(target_languages, list):
+                target_languages = [str(target_languages)]
+            target_languages_str = ", ".join(target_languages)
+
             # Build rows JSON for LLM
             rows_json = json.dumps([
                 {
@@ -548,12 +664,13 @@ class TranslationExecutor:
                 for idx, row in enumerate(batch_rows)
             ], indent=2)
 
-            # Build prompt
+            # Build prompt (use multi-language format)
             prompt = render_prompt(
                 DEFAULT_EXECUTION_PROMPT_TEMPLATE,
                 {
                     "source_language": job.source_dialect or "SQL",
-                    "target_language": job.target_language or job.target_dialect or "en",
+                    "target_language": target_languages_str,
+                    "target_languages": target_languages_str,
                     "source_dialect": job.source_dialect or "",
                     "target_dialect": job.target_dialect or "",
                     "translation_column": job.translation_column or "",
@@ -607,9 +724,9 @@ class TranslationExecutor:
                     self.db.add(record)
                 return {"successful": 0, "failed": len(batch_rows), "skipped": 0, "retries": retries}
 
-            # Parse LLM response
+            # Parse LLM response (multi-language aware)
             try:
-                translations = self._parse_llm_response(llm_response, len(batch_rows))
+                translations = self._parse_llm_response(llm_response, len(batch_rows), target_languages=target_languages)
             except ValueError as e:
                 # Parse failure — mark all rows as SKIPPED
                 logger.explore("LLM response parse failed", {
@@ -646,16 +763,20 @@ class TranslationExecutor:
 
             for row in batch_rows:
                 row_id = str(row.get("row_index", ""))
-                translation = translations.get(row_id)
+                translation_data = translations.get(row_id)
+                source_text = row.get("source_text", "")
+                detected_lang = "und"
+                if translation_data:
+                    detected_lang = translation_data.get("detected_source_language", "und")
 
-                if translation is None:
+                if translation_data is None:
                     # NULL translation — skip
                     skipped += 1
                     record = TranslationRecord(
                         id=str(uuid.uuid4()),
                         batch_id=batch_id,
                         run_id=run_id,
-                        source_sql=row.get("source_text", ""),
+                        source_sql=source_text,
                         target_sql="",
                         source_object_type="table_row",
                         source_object_id=row.get("row_index"),
@@ -667,14 +788,32 @@ class TranslationExecutor:
                     self.db.add(record)
                     continue
 
-                if translation.strip() == "":
-                    # Empty translation — skip
+                # Collect per-language translated values
+                per_lang_values: dict[str, str] = {}
+                has_any_translation = False
+
+                # First try multi-language format (per-language keys)
+                for lang_code in target_languages:
+                    lang_val = translation_data.get(lang_code)
+                    if lang_val is not None and str(lang_val).strip():
+                        per_lang_values[lang_code] = str(lang_val)
+                        has_any_translation = True
+
+                # Fallback to legacy single "translation" key format
+                if not has_any_translation:
+                    translation_text = translation_data.get("translation", "")
+                    if translation_text.strip():
+                        per_lang_values[target_languages[0]] = translation_text
+                        has_any_translation = True
+
+                if not has_any_translation:
+                    # Empty/all-empty translations — skip
                     skipped += 1
                     record = TranslationRecord(
                         id=str(uuid.uuid4()),
                         batch_id=batch_id,
                         run_id=run_id,
-                        source_sql=row.get("source_text", ""),
+                        source_sql=source_text,
                         target_sql="",
                         source_object_type="table_row",
                         source_object_id=row.get("row_index"),
@@ -686,13 +825,16 @@ class TranslationExecutor:
                     self.db.add(record)
                     continue
 
+                # Use the first language's value as the primary target_sql for backward compat
+                primary_translation = next(iter(per_lang_values.values()), "")
+
                 successful += 1
                 record = TranslationRecord(
                     id=str(uuid.uuid4()),
                     batch_id=batch_id,
                     run_id=run_id,
-                    source_sql=row.get("source_text", ""),
-                    target_sql=translation,
+                    source_sql=source_text,
+                    target_sql=primary_translation,
                     source_object_type="table_row",
                     source_object_id=row.get("row_index"),
                     source_object_name=row.get("source_object_name", ""),
@@ -701,6 +843,37 @@ class TranslationExecutor:
                 )
                 self.db.add(record)
 
+                # Create per-language entries — one TranslationLanguage per language_code
+                for lang_code in target_languages:
+                    lang_translation = per_lang_values.get(lang_code, "")
+
+                    # Source-as-reference: if detected source language matches this target,
+                    # store the original text verbatim (no translation needed)
+                    if detected_lang != "und" and str(lang_code).lower() == str(detected_lang).lower():
+                        lang_translation = source_text
+
+                    # Check for undetermined source language
+                    lang_needs_review = (detected_lang == "und")
+
+                    if lang_needs_review:
+                        logger.explore("undetected language", {
+                            "record_id": row_id,
+                            "language_code": lang_code,
+                            "source_text": source_text[:100],
+                        })
+
+                    lang_entry = TranslationLanguage(
+                        id=str(uuid.uuid4()),
+                        record_id=record.id,
+                        language_code=lang_code,
+                        source_language_detected=detected_lang,
+                        translated_value=lang_translation or "",
+                        final_value=lang_translation or "",
+                        status="translated",
+                        needs_review=lang_needs_review,
+                    )
+                    self.db.add(lang_entry)
+
             return {
                 "successful": successful,
                 "failed": failed,
@@ -806,11 +979,13 @@ class TranslationExecutor:
     # endregion _call_openai_compatible
 
     # region _parse_llm_response [TYPE Function]
-    # @PURPOSE: Parse LLM JSON response into dict of row_id -> translation.
+    # @PURPOSE: Parse LLM JSON response into dict of row_id -> per-language translations.
+    #   Supports multi-language format: {"row_id": 0, "detected_source_language": "fr", "ru": "текст", "en": "text"}
+    #   Backward-compat: old format {"row_id": 0, "translation": "text", "detected_source_language": "fr"}
     # @PRE: response_text is valid JSON with {"rows": [...]} structure.
-    # @POST: Returns dict mapping row_id to translation text.
+    # @POST: Returns dict mapping row_id to dict with 'detected_source_language' and per-language codes.
     @staticmethod
-    def _parse_llm_response(response_text: str, expected_count: int) -> dict[str, str]:
+    def _parse_llm_response(response_text: str, expected_count: int, target_languages: list[str] | None = None) -> dict[str, dict[str, str]]:
         with belief_scope("TranslationExecutor._parse_llm_response"):
             try:
                 data = json.loads(response_text)
@@ -830,15 +1005,41 @@ class TranslationExecutor:
             if not isinstance(rows, list):
                 raise ValueError("LLM response missing 'rows' array")
 
-            translations: dict[str, str] = {}
+            translations: dict[str, dict[str, str]] = {}
             for item in rows:
                 row_id = str(item.get("row_id", ""))
-                translation = item.get("translation")
-                if translation is None:
-                    # Skip NULL translations — they'll be handled by caller
+                if not row_id:
                     continue
-                if row_id:
-                    translations[row_id] = str(translation)
+
+                detected_lang = str(item.get("detected_source_language", "und")) if item.get("detected_source_language") else "und"
+                result: dict[str, str] = {
+                    "detected_source_language": detected_lang,
+                }
+
+                # Multi-language format: extract per-language keys
+                has_language_data = False
+                if target_languages:
+                    for lang_code in target_languages:
+                        lang_val = item.get(lang_code)
+                        if lang_val is not None and str(lang_val).strip():
+                            result[lang_code] = str(lang_val)
+                            has_language_data = True
+
+                # Fallback to legacy single "translation" key format
+                if not has_language_data:
+                    translation = item.get("translation")
+                    if translation is not None:
+                        result["translation"] = str(translation)
+                        has_language_data = True
+
+                if has_language_data:
+                    translations[row_id] = result
+
+            if len(translations) < expected_count:
+                logger.explore(
+                    f"LLM returned fewer translations expected={expected_count} "
+                    f"got={len(translations)}"
+                )
 
             return translations
     # endregion _parse_llm_response

backend/src/plugins/translate/metrics.py

@@ -21,6 +21,7 @@ from ...models.translate import (
     MetricSnapshot,
     TranslationEvent,
     TranslationRun,
+    TranslationRunLanguageStats,
     TranslationSchedule,
 )
 
@@ -121,17 +122,37 @@ class TranslationMetrics:
                 # indicates cutoff
                 pass
 
-            # Live events (<90 days) for token/cost
-            cutoff = datetime.now(UTC)
-            live_events = (
-                self.db.query(TranslationEvent)
-                .filter(
-                    TranslationEvent.job_id == job_id,
-                    TranslationEvent.event_type.in_(["TRANSLATION_PHASE_COMPLETED", "RUN_COMPLETED"]),
-                    TranslationEvent.created_at > cutoff,  # events newer than snapshot
-                )
+            # Per-language metrics from TranslationRunLanguageStats (live runs)
+            per_language: dict[str, dict[str, int | float]] = {}
+            lang_stats_rows = (
+                self.db.query(TranslationRunLanguageStats)
+                .join(TranslationRun, TranslationRunLanguageStats.run_id == TranslationRun.id)
+                .filter(TranslationRun.job_id == job_id)
                 .all()
             )
+            for ls in lang_stats_rows:
+                lang = ls.language_code
+                if lang not in per_language:
+                    per_language[lang] = {"tokens": 0, "cost": 0.0, "runs": 0, "translated_rows": 0}
+                per_language[lang]["tokens"] += ls.token_count or 0
+                per_language[lang]["cost"] += ls.estimated_cost or 0.0
+                per_language[lang]["runs"] += 1
+                per_language[lang]["translated_rows"] += ls.translated_rows or 0
+
+            # Merge in per-language data from MetricSnapshot (pruned period)
+            latest_snapshot = (
+                self.db.query(MetricSnapshot)
+                .filter(MetricSnapshot.job_id == job_id)
+                .order_by(MetricSnapshot.snapshot_date.desc())
+                .first()
+            )
+            if latest_snapshot and latest_snapshot.per_language_metrics:
+                for lang, snap_data in latest_snapshot.per_language_metrics.items():
+                    if lang not in per_language:
+                        per_language[lang] = {"tokens": 0, "cost": 0.0, "runs": 0, "translated_rows": 0}
+                    per_language[lang]["tokens"] += snap_data.get("cumulative_tokens", 0)
+                    per_language[lang]["cost"] += snap_data.get("cumulative_cost", 0.0)
+                    per_language[lang]["runs"] += snap_data.get("runs", 0)
 
             return {
                 "job_id": job_id,
@@ -148,6 +169,7 @@ class TranslationMetrics:
                 "avg_duration_ms": int(avg_duration) if avg_duration else None,
                 "last_run_at": last_run.created_at.isoformat() if last_run else None,
                 "next_scheduled_run": next_schedule.last_run_at.isoformat() if next_schedule and next_schedule.last_run_at else None,
+                "per_language_metrics": per_language,
             }
     # endregion get_job_metrics
 

backend/src/plugins/translate/orchestrator.py

@@ -30,9 +30,11 @@ from ...core.logger import belief_scope, logger
 from ...models.translate import (
     TranslationBatch,
     TranslationJob,
+    TranslationLanguage,
     TranslationPreviewSession,
     TranslationRecord,
     TranslationRun,
+    TranslationRunLanguageStats,
 )
 from .events import TranslationEventLog
 from .executor import TranslationExecutor
@@ -201,6 +203,27 @@ class TranslationOrchestrator:
                 created_by=self.current_user,
             )
 
+            # Initialize per-language stats
+            target_languages = job.target_languages or [job.target_language or job.target_dialect or "en"]
+            if not isinstance(target_languages, list):
+                target_languages = [str(target_languages)]
+            language_stats_map: dict[str, TranslationRunLanguageStats] = {}
+            for lang_code in target_languages:
+                lang_stat = TranslationRunLanguageStats(
+                    id=str(uuid.uuid4()),
+                    run_id=run.id,
+                    language_code=lang_code,
+                    total_rows=0,
+                    translated_rows=0,
+                    failed_rows=0,
+                    skipped_rows=0,
+                    token_count=0,
+                    estimated_cost=0.0,
+                )
+                self.db.add(lang_stat)
+                language_stats_map[lang_code] = lang_stat
+            self.db.flush()
+
             # Dispatch executor
             executor = TranslationExecutor(
                 self.db, self.config_manager, self.current_user,
@@ -227,6 +250,9 @@ class TranslationOrchestrator:
                 self.db.commit()
                 return run
 
+            # Aggregate per-language statistics after executor completes
+            self._update_language_stats(run.id, language_stats_map)
+
             # Record translation phase complete
             self.event_log.log_event(
                 job_id=job.id,
@@ -745,6 +771,25 @@ class TranslationOrchestrator:
             # Get event summary
             event_summary = self.event_log.get_run_event_summary(run_id)
 
+            # Get language stats
+            language_stats_entries = (
+                self.db.query(TranslationRunLanguageStats)
+                .filter(TranslationRunLanguageStats.run_id == run_id)
+                .all()
+            )
+            language_stats = [
+                {
+                    "language_code": ls.language_code,
+                    "total_rows": ls.total_rows or 0,
+                    "translated_rows": ls.translated_rows or 0,
+                    "failed_rows": ls.failed_rows or 0,
+                    "skipped_rows": ls.skipped_rows or 0,
+                    "token_count": ls.token_count or 0,
+                    "estimated_cost": ls.estimated_cost or 0.0,
+                }
+                for ls in language_stats_entries
+            ]
+
             return {
                 "id": run.id,
                 "job_id": run.job_id,
@@ -759,6 +804,7 @@ class TranslationOrchestrator:
                 "insert_status": run.insert_status,
                 "superset_execution_id": run.superset_execution_id,
                 "batch_count": batch_count,
+                "language_stats": language_stats,
                 "event_invariants": {
                     "has_run_started": event_summary["has_run_started"],
                     "terminal_event_count": event_summary["terminal_event_count"],
@@ -863,6 +909,80 @@ class TranslationOrchestrator:
             ]
     # endregion get_run_history
 
+    # region _update_language_stats [TYPE Function]
+    # @PURPOSE: Aggregate TranslationLanguage entries by language_code and update TranslationRunLanguageStats.
+    # @PRE: run_id and language_stats_map are valid. DB session is available.
+    # @POST: Language stats are updated with row counts and estimated tokens/cost.
+    # @SIDE_EFFECT: DB writes on language_stats objects.
+    def _update_language_stats(
+        self,
+        run_id: str,
+        language_stats_map: dict[str, TranslationRunLanguageStats],
+    ) -> None:
+        with belief_scope("TranslationOrchestrator._update_language_stats"):
+            # Get all records for this run to join with TranslationLanguage
+            records = (
+                self.db.query(TranslationRecord)
+                .filter(TranslationRecord.run_id == run_id)
+                .all()
+            )
+            record_ids = [r.id for r in records]
+
+            if not record_ids:
+                logger.reason("No records for language stats aggregation", {"run_id": run_id})
+                return
+
+            # Get all language entries for this run's records
+            lang_entries = (
+                self.db.query(TranslationLanguage)
+                .filter(TranslationLanguage.record_id.in_(record_ids))
+                .all()
+            )
+
+            # Aggregate by language_code
+            from collections import defaultdict
+            agg: dict[str, dict[str, int]] = defaultdict(lambda: {"total": 0, "translated": 0, "failed": 0, "skipped": 0})
+
+            for le in lang_entries:
+                code = le.language_code
+                agg[code]["total"] += 1
+                if le.status in ("translated", "approved", "edited"):
+                    agg[code]["translated"] += 1
+                elif le.status == "failed":
+                    agg[code]["failed"] += 1
+                elif le.status == "skipped":
+                    agg[code]["skipped"] += 1
+
+            # Estimate tokens: heuristic based on character count of translated values
+            total_chars = sum(
+                len(le.translated_value or "") for le in lang_entries if le.translated_value
+            )
+            total_tokens = max(1, total_chars // 4)  # ~4 chars per token
+            cost_per_token = 0.002 / 1000  # $0.002 per 1K tokens
+
+            # Update each language stat entry
+            for lang_code, lang_stat in language_stats_map.items():
+                data = agg.get(lang_code, {"total": 0, "translated": 0, "failed": 0, "skipped": 0})
+                lang_stat.total_rows = data["total"]
+                lang_stat.translated_rows = data["translated"]
+                lang_stat.failed_rows = data["failed"]
+                lang_stat.skipped_rows = data["skipped"]
+
+                # Proportional token split: share tokens across languages
+                num_langs = len(language_stats_map)
+                if num_langs > 0:
+                    lang_stat.token_count = total_tokens // num_langs
+                    lang_stat.estimated_cost = round((lang_stat.token_count / 1000) * cost_per_token, 6)
+
+            self.db.flush()
+
+            logger.reason("Language stats updated", {
+                "run_id": run_id,
+                "languages": list(language_stats_map.keys()),
+                "total_tokens_est": total_tokens,
+            })
+    # endregion _update_language_stats
+
     # region _compute_config_hash [TYPE Function]
     # @PURPOSE: Compute a hash of the job's current configuration for snapshot comparison.
     @staticmethod

backend/src/plugins/translate/preview.py

@@ -29,6 +29,7 @@ from ...core.superset_client import SupersetClient
 from ...models.translate import (
     TranslationJob,
     TranslationJobDictionary,
+    TranslationPreviewLanguage,
     TranslationPreviewRecord,
     TranslationPreviewSession,
 )
@@ -38,16 +39,19 @@ from .dictionary import DictionaryManager
 
 # #region DEFAULT_EXECUTION_PROMPT_TEMPLATE [TYPE Constant]
 # @BRIEF Default prompt template for batch LLM translation execution (no context columns — faster).
+# Supports both single-language and multi-language modes via {target_languages} placeholder.
 DEFAULT_EXECUTION_PROMPT_TEMPLATE: str = (
-    "Translate the following database content from {source_language} to {target_language}.\n\n"
+    "Translate the following database content from {source_language} to the following language(s): {target_languages}.\n\n"
     "Source dialect: {source_dialect}\n"
-    "Target dialect: {target_dialect}\n"
+    "Target dialect(s): {target_dialect}\n"
     "Column to translate: {translation_column}\n\n"
     "{dictionary_section}"
-    "For each row, provide an accurate translation of the text.\n\n"
+    "For each row, provide an accurate translation of the text into each target language.\n\n"
     "Rows to translate:\n{rows_json}\n\n"
     "Respond with a JSON object in this exact format:\n"
-    '{{"rows": [{{"row_id": "<row_index>", "translation": "<translated_text>"}}]}}\n'
+    '{{"rows": [{{"row_id": "<row_index>", "detected_source_language": "<bcp47_or_und>", "<language_code_1>": "<translation_in_lang_1>", "<language_code_2>": "<translation_in_lang_2>"}}]}}\n'
+    "For each row, return the detected source language as a BCP-47 tag (e.g. 'en', 'ru', 'fr'), or 'und' if uncertain.\n"
+    "Include a separate key for EACH target language code with the translated text in that language.\n"
     "Each row_id must match the index provided. Return exactly {row_count} entries."
 )
 # #endregion DEFAULT_EXECUTION_PROMPT_TEMPLATE
@@ -56,16 +60,18 @@ DEFAULT_EXECUTION_PROMPT_TEMPLATE: str = (
 # #region DEFAULT_PREVIEW_PROMPT_TEMPLATE [TYPE Constant]
 # @BRIEF Default prompt template for LLM translation preview.
 DEFAULT_PREVIEW_PROMPT_TEMPLATE: str = (
-    "Translate the following database content from {source_language} to {target_language}.\n\n"
+    "Translate the following database content.\n\n"
     "Source dialect: {source_dialect}\n"
-    "Target dialect: {target_dialect}\n"
     "Column to translate: {translation_column}\n\n"
     "{dictionary_section}"
-    "For each row, provide an accurate translation of the '{translation_column}' value.\n"
+    "Translate to the following languages: {target_languages}\n\n"
+    "For each row, provide an accurate translation of the '{translation_column}' value into each language.\n"
     "Consider the context columns when determining the meaning of the text.\n\n"
     "Rows to translate:\n{rows_json}\n\n"
     "Respond with a JSON object in this exact format:\n"
-    '{{"rows": [{{"row_id": "<row_index>", "translation": "<translated_text>"}}]}}\n'
+    '{{"rows": [{{"row_id": "<row_index>", "detected_source_language": "<bcp47_or_und>", "language_code_1": "<translation_in_lang_1>", "language_code_2": "<translation_in_lang_2>"}}]}}\n'
+    "For each row, return the detected source language as a BCP-47 tag (e.g. 'en', 'ru', 'fr'), or 'und' if uncertain.\n"
+    "Include a separate key for EACH target language code with the translated text in that language.\n"
     "Each row_id must match the index provided. Return exactly {row_count} entries."
 )
 # #endregion DEFAULT_PREVIEW_PROMPT_TEMPLATE
@@ -80,7 +86,9 @@ class TokenEstimator:
 
     CHARS_PER_TOKEN_ESTIMATE: float = 4.0
     OUTPUT_TOKENS_PER_ROW_ESTIMATE: int = 50
+    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
 
     # region estimate_prompt_tokens [TYPE Function]
     # @PURPOSE: Estimate token count for a prompt string.
@@ -94,12 +102,12 @@ class TokenEstimator:
     # endregion estimate_prompt_tokens
 
     # region estimate_output_tokens [TYPE Function]
-    # @PURPOSE: Estimate output token count for translating N rows.
-    # @PRE: row_count >= 0.
+    # @PURPOSE: Estimate output token count for translating N rows across N languages.
+    # @PRE: row_count >= 0, num_languages >= 1.
     # @POST: Returns estimated output token count.
     @staticmethod
-    def estimate_output_tokens(row_count: int) -> int:
-        return row_count * TokenEstimator.OUTPUT_TOKENS_PER_ROW_ESTIMATE
+    def estimate_output_tokens(row_count: int, num_languages: int = 1) -> int:
+        return int(row_count * num_languages * TokenEstimator.OUTPUT_TOKENS_PER_ROW_ESTIMATE * TokenEstimator.MULTI_LANG_FACTOR)
     # endregion estimate_output_tokens
 
     # region estimate_cost [TYPE Function]
@@ -112,6 +120,20 @@ class TokenEstimator:
         return round((total_tokens / 1000) * rate, 6)
     # endregion estimate_cost
 
+    # region check_cost_warning [TYPE Function]
+    # @PURPOSE: Generate cost warning for large previews.
+    # @PRE: sample_size > 0, num_languages >= 1.
+    # @POST: Returns warning string or None.
+    @staticmethod
+    def check_cost_warning(sample_size: int, num_languages: int, estimated_tokens: int, estimated_cost: float) -> str | None:
+        if sample_size > TokenEstimator.COST_WARNING_THRESHOLD:
+            return (
+                f"Large preview — estimated {estimated_tokens} tokens, ~${estimated_cost:.4f} cost "
+                f"(across {num_languages} languages)"
+            )
+        return None
+    # endregion check_cost_warning
+
 
 # #endregion TokenEstimator
 
@@ -134,10 +156,10 @@ class TranslationPreview:
         self.current_user = current_user
 
     # region preview_rows [TYPE Function]
-    # @PURPOSE: Fetch sample rows from Superset dataset, send to LLM for translation, create preview session with records.
+    # @PURPOSE: Fetch sample rows from Superset dataset, send to LLM for multi-language translation, create preview session with per-language records.
     # @PRE: job_id exists and job has source_datasource_id, translation_column configured.
-    # @POST: Returns TranslationPreviewResponse with records, cost estimation, and persistent session.
-    # @SIDE_EFFECT: Fetches data from Superset; calls LLM; creates TranslationPreviewSession and TranslationPreviewRecord rows.
+    # @POST: Returns TranslationPreviewResponse with per-language records, cost estimation, and persistent session.
+    # @SIDE_EFFECT: Fetches data from Superset; calls LLM; creates TranslationPreviewSession and TranslationPreviewRecord + TranslationPreviewLanguage rows.
     def preview_rows(
         self,
         job_id: str,
@@ -157,6 +179,12 @@ class TranslationPreview:
             if not job.translation_column:
                 raise ValueError("Job must have a translation column configured for preview")
 
+            # Resolve target languages: prefer target_languages list, fallback to single target_language
+            target_languages = job.target_languages or [job.target_language or job.target_dialect or "en"]
+            if not isinstance(target_languages, list):
+                target_languages = [str(target_languages)]
+            num_languages = len(target_languages)
+
             # 2. Compute config hash and dict snapshot hash
             config_hash = self._compute_config_hash(job)
             dict_snapshot_hash = self._compute_dict_snapshot_hash(job_id)
@@ -224,40 +252,48 @@ class TranslationPreview:
                     + "\n\n"
                 )
 
-            # 6. Build LLM prompt
+            # 6. Build LLM prompt with multi-language instructions
             rows_json = json.dumps([
                 {"row_id": str(m["row_index"]), "text": m["source_text"], "context": m["context_data"]}
                 for m in row_meta
             ], indent=2)
 
+            target_languages_str = ", ".join(target_languages)
             template = prompt_template or DEFAULT_PREVIEW_PROMPT_TEMPLATE
             prompt = render_prompt(template, {
                 "source_language": job.source_dialect or "SQL",
-                "target_language": job.target_language or job.target_dialect or "en",
+                "target_language": target_languages_str,
                 "source_dialect": job.source_dialect or "",
-                "target_dialect": job.target_dialect or "",
+                "target_languages": target_languages_str,
                 "translation_column": job.translation_column or "",
                 "dictionary_section": dictionary_section,
                 "rows_json": rows_json,
                 "row_count": str(actual_row_count),
             })
 
-            # 7. Estimate tokens/cost for sample and full dataset
+            # 7. Estimate tokens/cost for sample with multi-language factor
             sample_prompt_tokens = TokenEstimator.estimate_prompt_tokens(prompt)
-            sample_output_tokens = TokenEstimator.estimate_output_tokens(actual_row_count)
+            sample_output_tokens = TokenEstimator.estimate_output_tokens(actual_row_count, num_languages)
             sample_total_tokens = sample_prompt_tokens + sample_output_tokens
             sample_cost = TokenEstimator.estimate_cost(sample_total_tokens)
 
-            # Estimate full dataset cost (if we knew total rows)
+            # Estimate full dataset cost (extrapolated)
+            total_est_rows = sample_size * 10  # rough extrapolation
             total_est_tokens = TokenEstimator.estimate_prompt_tokens(
-                prompt.replace(str(actual_row_count), "{total}")
-            ) + TokenEstimator.estimate_output_tokens(sample_size * 10)  # rough extrapolation
+                prompt.replace(str(actual_row_count), str(total_est_rows))
+            ) + TokenEstimator.estimate_output_tokens(total_est_rows, num_languages)
             total_est_cost = TokenEstimator.estimate_cost(total_est_tokens)
 
+            # Cost warning for large previews
+            cost_warning = TokenEstimator.check_cost_warning(
+                sample_size, num_languages, sample_total_tokens, sample_cost
+            )
+
             # 8. Call LLM
             logger.reason("Calling LLM for preview translation", {
                 "provider_id": job.provider_id,
                 "row_count": actual_row_count,
+                "num_languages": num_languages,
                 "estimated_tokens": sample_total_tokens,
             })
             llm_response = self._call_llm(
@@ -265,8 +301,10 @@ class TranslationPreview:
                 prompt=prompt,
             )
 
-            # 9. Parse LLM response
-            translations = self._parse_llm_response(llm_response, actual_row_count)
+            # 9. Parse LLM response (multi-language)
+            translations = self._parse_llm_response(
+                llm_response, actual_row_count, target_languages=target_languages
+            )
 
             # 10. Create preview session
             session = TranslationPreviewSession(
@@ -280,14 +318,50 @@ class TranslationPreview:
             self.db.add(session)
             self.db.flush()
 
-            # 11. Create preview records
+            # 11. Create preview records with per-language entries
             records = []
             for meta in row_meta:
                 idx = meta["row_index"]
-                translation = translations.get(str(idx), "")
-                is_rejected = False
-                status = "PENDING"
-                feedback = None
+                source_text = meta["source_text"]
+                translation_data = translations.get(str(idx), {})
+                detected_lang = translation_data.get("detected_source_language", "und") if isinstance(translation_data, dict) else "und"
+
+                # Extract per-language translations
+                lang_entries: list[TranslationPreviewLanguage] = []
+                for lang_code in target_languages:
+                    # Get translation for this language from LLM response
+                    lang_translation = translation_data.get(lang_code) if isinstance(translation_data, dict) else None
+
+                    # Fallback: if no per-language data, use the single "translation" key
+                    if not lang_translation:
+                        if isinstance(translation_data, dict):
+                            lang_translation = translation_data.get("translation", "")
+                        elif isinstance(translation_data, str):
+                            lang_translation = translation_data
+                        else:
+                            lang_translation = ""
+
+                    # Source-as-reference: if this language code matches source, use original text
+                    if str(lang_code).lower() == str(detected_lang).lower() and detected_lang != "und":
+                        # Use source text as reference (no translation needed for source language)
+                        lang_translation = source_text
+
+                    lang_needs_review = detected_lang == "und"
+
+                    lang_entry = TranslationPreviewLanguage(
+                        id=str(uuid.uuid4()),
+                        preview_record_id="",  # Will set after record is created
+                        language_code=lang_code,
+                        source_language_detected=detected_lang,
+                        translated_value=str(lang_translation),
+                        final_value=str(lang_translation),
+                        status="pending",
+                        needs_review=lang_needs_review,
+                    )
+                    lang_entries.append(lang_entry)
+
+                # Determine overall status and needs_review for the record
+                overall_needs_review = any(le.needs_review for le in lang_entries)
 
                 # Extract source_data: store original row key columns for upsert matching
                 source_row = meta.get("source_row", {})
@@ -299,24 +373,41 @@ class TranslationPreview:
                         if k in source_row
                     }
                 elif source_row:
-                    # No key columns configured — store the full row as fallback
                     source_data = dict(source_row)
 
+                # Use first language's translation as target_sql for backward compat
+                primary_translation = lang_entries[0].translated_value if lang_entries else ""
+
                 record = TranslationPreviewRecord(
                     id=str(uuid.uuid4()),
                     session_id=session.id,
-                    source_sql=meta["source_text"],
-                    target_sql=translation,
+                    source_sql=source_text,
+                    target_sql=primary_translation,
                     source_object_type="table_row",
                     source_object_id=str(idx),
                     source_object_name=f"Row {idx + 1}",
                     source_data=source_data,
-                    status=status,
-                    feedback=feedback,
+                    status="PENDING",
+                    feedback=None,
                     created_at=datetime.now(UTC),
                 )
                 self.db.add(record)
                 self.db.flush()
+
+                # Now set the preview_record_id for all lang entries and persist
+                serialized_langs = []
+                for le in lang_entries:
+                    le.preview_record_id = record.id
+                    self.db.add(le)
+                    serialized_langs.append({
+                        "language_code": le.language_code,
+                        "source_language_detected": le.source_language_detected,
+                        "translated_value": le.translated_value,
+                        "final_value": le.final_value,
+                        "status": le.status,
+                        "needs_review": le.needs_review,
+                    })
+
                 records.append({
                     "id": record.id,
                     "source_sql": record.source_sql,
@@ -326,6 +417,9 @@ class TranslationPreview:
                     "source_object_name": record.source_object_name,
                     "status": record.status,
                     "feedback": record.feedback,
+                    "source_language_detected": detected_lang,
+                    "needs_review": overall_needs_review,
+                    "languages": serialized_langs,
                 })
 
             self.db.commit()
@@ -338,15 +432,18 @@ class TranslationPreview:
                 "created_at": session.created_at.isoformat(),
                 "expires_at": session.expires_at.isoformat() if session.expires_at else None,
                 "records": records,
+                "target_languages": target_languages,
                 "cost_estimate": {
                     "sample_size": actual_row_count,
+                    "num_languages": num_languages,
                     "sample_prompt_tokens": sample_prompt_tokens,
                     "sample_output_tokens": sample_output_tokens,
                     "sample_total_tokens": sample_total_tokens,
                     "sample_cost": sample_cost,
-                    "estimated_total_rows": actual_row_count * 10,
+                    "estimated_total_rows": total_est_rows,
                     "estimated_tokens": total_est_tokens,
                     "estimated_cost": total_est_cost,
+                    "warning": cost_warning,
                 },
                 "config_hash": config_hash,
                 "dict_snapshot_hash": dict_snapshot_hash,
@@ -355,15 +452,16 @@ class TranslationPreview:
             logger.reflect("Preview completed", {
                 "session_id": session.id,
                 "row_count": actual_row_count,
+                "num_languages": num_languages,
                 "sample_cost": sample_cost,
             })
             return result
     # endregion preview_rows
 
     # region update_preview_row [TYPE Function]
-    # @PURPOSE: Approve, edit, or reject an individual preview row.
+    # @PURPOSE: Approve, edit, or reject an individual preview row (optionally per language).
     # @PRE: session_id and row_id exist, session is ACTIVE.
-    # @POST: PreviewRecord status is updated.
+    # @POST: PreviewRecord status is updated. If language_code provided, only that TranslationPreviewLanguage is updated.
     def update_preview_row(
         self,
         job_id: str,
@@ -371,6 +469,7 @@ class TranslationPreview:
         action: str,
         translation: str | None = None,
         feedback: str | None = None,
+        language_code: str | None = None,
     ) -> dict[str, Any]:
         with belief_scope("TranslationPreview.update_preview_row"):
             # Find the active session for this job
@@ -397,16 +496,76 @@ class TranslationPreview:
             if not record:
                 raise ValueError(f"Preview record '{row_id}' not found in active session")
 
-            if action == "approve":
-                record.status = "APPROVED"
-            elif action == "reject":
-                record.status = "REJECTED"
-            elif action == "edit":
-                record.status = "APPROVED"
-                if translation is not None:
-                    record.target_sql = translation
+            # If language_code specified, operate on the specific TranslationPreviewLanguage
+            if language_code:
+                lang_entry = (
+                    self.db.query(TranslationPreviewLanguage)
+                    .filter(
+                        TranslationPreviewLanguage.preview_record_id == row_id,
+                        TranslationPreviewLanguage.language_code == language_code,
+                    )
+                    .first()
+                )
+                if not lang_entry:
+                    raise ValueError(f"Language entry '{language_code}' not found for record '{row_id}'")
+
+                if action == "approve":
+                    lang_entry.status = "approved"
+                elif action == "reject":
+                    lang_entry.status = "rejected"
+                elif action == "edit":
+                    lang_entry.status = "edited"
+                    if translation is not None:
+                        lang_entry.translated_value = translation
+                        lang_entry.final_value = translation
+                        lang_entry.user_edit = translation
+                else:
+                    raise ValueError(f"Invalid action '{action}'. Use 'approve', 'reject', or 'edit'.")
+
+                # Update record-level status based on all language entries
+                all_langs = (
+                    self.db.query(TranslationPreviewLanguage)
+                    .filter(TranslationPreviewLanguage.preview_record_id == row_id)
+                    .all()
+                )
+                all_approved = all(le.status in ("approved", "edited") for le in all_langs)
+                any_rejected = any(le.status == "rejected" for le in all_langs)
+
+                if all_approved:
+                    record.status = "APPROVED"
+                elif any_rejected and not all_langs:
+                    record.status = "REJECTED"
+
+                # Update target_sql to reflect primary (first) language edit
+                if action == "edit" and translation is not None and all_langs:
+                    record.target_sql = all_langs[0].final_value or all_langs[0].translated_value
+
             else:
-                raise ValueError(f"Invalid action '{action}'. Use 'approve', 'reject', or 'edit'.")
+                # Legacy behavior: operate on whole record (all languages)
+                if action == "approve":
+                    record.status = "APPROVED"
+                    # Approve all pending language entries
+                    for lang_entry in record.languages:
+                        if lang_entry.status == "pending":
+                            lang_entry.status = "approved"
+                elif action == "reject":
+                    record.status = "REJECTED"
+                    for lang_entry in record.languages:
+                        if lang_entry.status == "pending":
+                            lang_entry.status = "rejected"
+                elif action == "edit":
+                    record.status = "APPROVED"
+                    if translation is not None:
+                        record.target_sql = translation
+                        # Update primary (first) language entry
+                        if record.languages:
+                            first_lang = record.languages[0]
+                            first_lang.translated_value = translation
+                            first_lang.final_value = translation
+                            first_lang.user_edit = translation
+                            first_lang.status = "edited"
+                else:
+                    raise ValueError(f"Invalid action '{action}'. Use 'approve', 'reject', or 'edit'.")
 
             if feedback is not None:
                 record.feedback = feedback
@@ -418,21 +577,37 @@ class TranslationPreview:
                 "row_id": row_id,
                 "session_id": session.id,
                 "status": record.status,
+                "language_code": language_code,
             })
 
+            # Build response with per-language data
+            lang_responses = []
+            if record.languages:
+                for le in record.languages:
+                    lang_responses.append({
+                        "language_code": le.language_code,
+                        "source_language_detected": le.source_language_detected,
+                        "translated_value": le.translated_value,
+                        "final_value": le.final_value,
+                        "user_edit": le.user_edit,
+                        "status": le.status,
+                        "needs_review": le.needs_review,
+                    })
+
             return {
                 "id": record.id,
                 "source_sql": record.source_sql,
                 "target_sql": record.target_sql,
                 "status": record.status,
                 "feedback": record.feedback,
+                "languages": lang_responses,
             }
     # endregion update_preview_row
 
     # region accept_preview_session [TYPE Function]
     # @PURPOSE: Mark a preview session as accepted, which gates full execution.
     # @PRE: job_id has an ACTIVE preview session.
-    # @POST: Session status changes to APPLIED.
+    # @POST: Session status changes to APPLIED. User edits are persisted for carry-forward.
     # @SIDE_EFFECT: Future full execution calls will check for accepted session.
     def accept_preview_session(self, job_id: str) -> dict[str, Any]:
         with belief_scope("TranslationPreview.accept_preview_session"):
@@ -464,6 +639,12 @@ class TranslationPreview:
                 .all()
             )
 
+            # Resolve target languages
+            job = self.db.query(TranslationJob).filter(TranslationJob.id == job_id).first()
+            target_languages = job.target_languages or [job.target_language or job.target_dialect or "en"] if job else ["en"]
+            if not isinstance(target_languages, list):
+                target_languages = [str(target_languages)]
+
             return {
                 "id": session.id,
                 "job_id": job_id,
@@ -471,6 +652,7 @@ class TranslationPreview:
                 "created_by": session.created_by,
                 "created_at": session.created_at.isoformat(),
                 "expires_at": session.expires_at.isoformat() if session.expires_at else None,
+                "target_languages": target_languages,
                 "records": [
                     {
                         "id": r.id,
@@ -478,6 +660,28 @@ class TranslationPreview:
                         "target_sql": r.target_sql,
                         "status": r.status,
                         "feedback": r.feedback,
+                        "source_language_detected": (
+                            r.languages[0].source_language_detected
+                            if r.languages and r.languages[0].source_language_detected
+                            else None
+                        ),
+                        "needs_review": (
+                            r.languages[0].needs_review
+                            if r.languages
+                            else False
+                        ),
+                        "languages": [
+                            {
+                                "language_code": le.language_code,
+                                "source_language_detected": le.source_language_detected,
+                                "translated_value": le.translated_value,
+                                "final_value": le.final_value,
+                                "user_edit": le.user_edit,
+                                "status": le.status,
+                                "needs_review": le.needs_review,
+                            }
+                            for le in (r.languages or [])
+                        ] if r.languages else [],
                     }
                     for r in records
                 ],
@@ -487,7 +691,7 @@ class TranslationPreview:
     # region get_preview_session [TYPE Function]
     # @PURPOSE: Get the latest preview session for a job with its records.
     # @PRE: job_id exists.
-    # @POST: Returns session data with records or raises ValueError.
+    # @POST: Returns session data with per-language records or raises ValueError.
     def get_preview_session(self, job_id: str) -> dict[str, Any]:
         with belief_scope("TranslationPreview.get_preview_session"):
             session = (
@@ -505,6 +709,12 @@ class TranslationPreview:
                 .all()
             )
 
+            # Resolve target languages
+            job = self.db.query(TranslationJob).filter(TranslationJob.id == job_id).first()
+            target_languages = job.target_languages or [job.target_language or job.target_dialect or "en"] if job else ["en"]
+            if not isinstance(target_languages, list):
+                target_languages = [str(target_languages)]
+
             return {
                 "id": session.id,
                 "job_id": job_id,
@@ -512,6 +722,7 @@ class TranslationPreview:
                 "created_by": session.created_by,
                 "created_at": session.created_at.isoformat(),
                 "expires_at": session.expires_at.isoformat() if session.expires_at else None,
+                "target_languages": target_languages,
                 "records": [
                     {
                         "id": r.id,
@@ -522,6 +733,28 @@ class TranslationPreview:
                         "source_object_name": r.source_object_name,
                         "status": r.status,
                         "feedback": r.feedback,
+                        "source_language_detected": (
+                            r.languages[0].source_language_detected
+                            if r.languages and r.languages[0].source_language_detected
+                            else None
+                        ),
+                        "needs_review": (
+                            r.languages[0].needs_review
+                            if r.languages
+                            else False
+                        ),
+                        "languages": [
+                            {
+                                "language_code": le.language_code,
+                                "source_language_detected": le.source_language_detected,
+                                "translated_value": le.translated_value,
+                                "final_value": le.final_value,
+                                "user_edit": le.user_edit,
+                                "status": le.status,
+                                "needs_review": le.needs_review,
+                            }
+                            for le in (r.languages or [])
+                        ] if r.languages else [],
                     }
                     for r in records
                 ],
@@ -752,11 +985,11 @@ class TranslationPreview:
     # endregion _call_openai_compatible
 
     # region _parse_llm_response [TYPE Function]
-    # @PURPOSE: Parse the LLM JSON response into a dict of row_id -> translation.
+    # @PURPOSE: Parse the LLM JSON response into a dict of row_id -> per-language translations.
     # @PRE: response_text is valid JSON with {"rows": [...]} structure.
-    # @POST: Returns dict mapping string row_id to translation text.
+    # @POST: Returns dict mapping row_id to dict with 'detected_source_language' and per-language keys.
     @staticmethod
-    def _parse_llm_response(response_text: str, expected_count: int) -> dict[str, str]:
+    def _parse_llm_response(response_text: str, expected_count: int, target_languages: list[str] | None = None) -> dict[str, dict[str, str]]:
         with belief_scope("TranslationPreview._parse_llm_response"):
             logger.reason(f"Raw LLM response length={len(response_text)} preview={response_text[:500]}")
 
@@ -779,12 +1012,37 @@ class TranslationPreview:
                 logger.explore(f"LLM response has no 'rows' array, keys={list(data.keys())} text_preview={response_text[:300]}")
                 raise ValueError("LLM response missing 'rows' array")
 
-            translations: dict[str, str] = {}
+            translations: dict[str, dict[str, str]] = {}
             for item in rows:
                 row_id = str(item.get("row_id", ""))
-                translation = str(item.get("translation", ""))
-                if row_id:
-                    translations[row_id] = translation
+                if not row_id:
+                    continue
+
+                detected_lang = str(item.get("detected_source_language", "und")) if item.get("detected_source_language") else "und"
+                result: dict[str, str] = {
+                    "detected_source_language": detected_lang,
+                }
+
+                # Multi-language format: extract per-language keys
+                # Keys that are NOT row_id, detected_source_language, or translation are language codes
+                has_language_data = False
+                if target_languages:
+                    for lang_code in target_languages:
+                        lang_val = item.get(lang_code)
+                        if lang_val is not None and str(lang_val).strip():
+                            result[lang_code] = str(lang_val)
+                            has_language_data = True
+
+                # Fallback to old format (single "translation" key)
+                if not has_language_data:
+                    translation = item.get("translation")
+                    if translation is not None:
+                        result["translation"] = str(translation)
+                    else:
+                        # Skip rows with no translation data at all
+                        continue
+
+                translations[row_id] = result
 
             if len(translations) < expected_count:
                 logger.explore(

backend/src/plugins/translate/prompt_builder.py

@@ -0,0 +1,152 @@
+# #region ContextAwarePromptBuilder [C:2] [TYPE Module] [SEMANTICS translate, prompt, context, dictionary]
+# @BRIEF Pure-function prompt builder that enhances dictionary entries with context annotations.
+# @LAYER: Domain
+# @RELATION DEPENDS_ON -> [DictionaryEntry:Class]
+# @RATIONALE: Pure functions only — no I/O, no DB access. Separated from executor for testability.
+# @REJECTED: Embedding context inline in the executor would make it untestable without mocking DB.
+#
+# Typical workflow:
+#   1. Call build_context_entries(dictionary_entries, row_context) to get annotated, prioritized entries
+#   2. Each entry is rendered via render_entry() with optional priority flag
+#   3. Jaccard similarity >= 0.5 triggers priority flagging
+
+import json
+from typing import Any
+
+# #region ContextAwarePromptBuilder [C:2] [TYPE Class]
+# @BRIEF Build LLM prompts with context-aware dictionary entries and similarity-based priority.
+
+class ContextAwarePromptBuilder:
+    """Build LLM prompts with context-aware dictionary entries.
+
+    Pure function — no I/O, no DB access.
+    """
+
+    @staticmethod
+    def render_entry(entry: Any, priority: bool = False, row_context: dict | None = None) -> str:
+        """Render a dictionary entry for the LLM prompt.
+
+        Args:
+            entry: DictionaryEntry-like object (must have source_term, target_term, has_context,
+                   context_data, usage_notes attrs or dict keys).
+            priority: Whether this entry should be flagged as high priority.
+            row_context: Optional row context dict (unused in rendering, kept for API symmetry).
+
+        Returns:
+            Rendered prompt line string.
+        """
+        # Support both object attrs and dict access
+        if isinstance(entry, dict):
+            source_term = entry.get("source_term", "")
+            target_term = entry.get("target_term", "")
+            has_context = entry.get("has_context", False)
+            context_data = entry.get("context_data")
+            usage_notes = entry.get("usage_notes")
+        else:
+            source_term = getattr(entry, "source_term", "")
+            target_term = getattr(entry, "target_term", "")
+            has_context = getattr(entry, "has_context", False)
+            context_data = getattr(entry, "context_data", None)
+            usage_notes = getattr(entry, "usage_notes", None)
+
+        # Build base line
+        line = f'"{source_term}" -> "{target_term}"'
+
+        # Add context annotation if present
+        if has_context and context_data:
+            context_items = []
+            if isinstance(context_data, dict):
+                context_items = [f"{k}={v}" for k, v in context_data.items()]
+            elif isinstance(context_data, str):
+                try:
+                    parsed = json.loads(context_data)
+                    if isinstance(parsed, dict):
+                        context_items = [f"{k}={v}" for k, v in parsed.items()]
+                    else:
+                        context_items = [str(context_data)]
+                except (json.JSONDecodeError, TypeError):
+                    context_items = [str(context_data)]
+
+            context_str = ", ".join(context_items)
+
+            # Truncate if too long (500 tokens ≈ 2000 chars)
+            if len(context_str) > 2000:
+                context_str = context_str[:1997] + "...[truncated]"
+
+            line = f'"{source_term}" (context: {context_str}) -> "{target_term}"'
+
+        # Add usage notes
+        if usage_notes:
+            notes = str(usage_notes)[:200]  # cap at 200 chars
+            line += f"  # Usage: {notes}"
+
+        # Add priority prefix
+        if priority:
+            line = f"# PRIORITY (context match) — {line}"
+
+        return line
+
+    @staticmethod
+    def compute_context_similarity(entry_context: dict | None, row_context: dict | None) -> float:
+        """Jaccard similarity between entry context and row context. Returns 0.0-1.0.
+
+        Compares the sets of lowercased string values from both contexts.
+        Returns 1.0 for identical contexts, 0.0 for disjoint or missing.
+        """
+        if not entry_context or not row_context:
+            return 0.0
+
+        entry_vals = set(str(v).lower() for v in entry_context.values() if v is not None)
+        row_vals = set(str(v).lower() for v in row_context.values() if v is not None)
+
+        if not entry_vals or not row_vals:
+            return 0.0
+
+        intersection = entry_vals & row_vals
+        union = entry_vals | row_vals
+        return len(intersection) / len(union)
+
+    @staticmethod
+    def build_context_entries(
+        dictionary_entries: list[Any],
+        row_context: dict | None = None,
+    ) -> list[str]:
+        """Build prioritized dictionary entry list with context annotations.
+
+        Args:
+            dictionary_entries: List of DictionaryEntry-like objects or dicts.
+            row_context: Optional dict of current row's context columns.
+
+        Returns:
+            List of rendered prompt strings, sorted with priority entries first.
+        """
+        results: list[tuple[Any, bool]] = []
+
+        for entry in dictionary_entries:
+            priority = False
+            if row_context:
+                # Extract entry context_data
+                if isinstance(entry, dict):
+                    entry_context = entry.get("context_data")
+                else:
+                    entry_context = getattr(entry, "context_data", None)
+
+                if entry_context:
+                    similarity = ContextAwarePromptBuilder.compute_context_similarity(
+                        entry_context, row_context
+                    )
+                    priority = similarity >= 0.5
+
+            results.append((entry, priority))
+
+        # Sort: priority first, then non-priority
+        results.sort(key=lambda x: (not x[1]))
+
+        return [
+            ContextAwarePromptBuilder.render_entry(entry, priority, row_context)
+            for entry, priority in results
+        ]
+# #endregion ContextAwarePromptBuilder
+
+
+# #endregion ContextAwarePromptBuilder

backend/src/plugins/translate/service.py

@@ -10,6 +10,7 @@
 # @RATIONALE: Snapshot isolation — in-progress runs use config snapshot; config edits affect future runs only.
 # @REJECTED: Invalidating in-progress runs on config edit would break scheduled run continuity.
 
+import re
 import uuid
 from datetime import UTC, datetime
 from typing import Any
@@ -27,6 +28,7 @@ from ...schemas.translate import (
     TranslateJobResponse,
     TranslateJobUpdate,
 )
+from .dictionary import _validate_bcp47
 
 # Supported database dialects for translation
 SUPPORTED_DIALECTS = {
@@ -227,6 +229,16 @@ class TranslateJobService:
                     logger.warning(f"[TranslateJobService] Dialect detection failed: {e}")
                     dialect = payload.source_dialect
 
+        # Resolve target_languages: accept new format (list) or legacy single language
+        target_languages = payload.target_languages
+        if not target_languages and payload.target_language:
+            target_languages = [payload.target_language]
+        if target_languages:
+            if not isinstance(target_languages, list):
+                target_languages = [str(target_languages)]
+            for lang in target_languages:
+                _validate_bcp47(lang, "target_languages")
+
         # Build job instance
         job = TranslationJob(
             id=str(uuid.uuid4()),
@@ -245,6 +257,7 @@ class TranslateJobService:
             target_column=payload.target_column,
             context_columns=payload.context_columns or [],
             target_language=payload.target_language,
+            target_languages=target_languages,
             provider_id=payload.provider_id,
             batch_size=payload.batch_size,
             upsert_strategy=payload.upsert_strategy,
@@ -284,6 +297,22 @@ class TranslateJobService:
         update_data = payload.model_dump(exclude_unset=True)
         dict_ids = update_data.pop("dictionary_ids", None)
 
+        # Backward compat: if only target_language is set (old API), wrap into target_languages
+        if "target_language" in update_data and "target_languages" not in update_data:
+            target_languages = [update_data["target_language"]] if update_data["target_language"] else []
+            update_data["target_languages"] = target_languages
+        elif "target_languages" in update_data and "target_language" not in update_data:
+            # Keep deprecated field in sync
+            if update_data["target_languages"]:
+                update_data["target_language"] = update_data["target_languages"][0]
+
+        # Validate BCP-47 for target_languages
+        if update_data.get("target_languages"):
+            if not isinstance(update_data["target_languages"], list):
+                update_data["target_languages"] = [str(update_data["target_languages"])]
+            for lang in update_data["target_languages"]:
+                _validate_bcp47(lang, "target_languages")
+
         for field, value in update_data.items():
             if hasattr(job, field):
                 setattr(job, field, value)
@@ -366,6 +395,7 @@ class TranslateJobService:
             target_column=source.target_column,
             context_columns=source.context_columns,
             target_language=source.target_language,
+            target_languages=source.target_languages,
             provider_id=source.provider_id,
             batch_size=source.batch_size,
             upsert_strategy=source.upsert_strategy,
@@ -468,6 +498,7 @@ def job_to_response(job: TranslationJob, dict_ids: list[str] | None = None) -> T
         target_column=job.target_column,
         context_columns=job.context_columns or [],
         target_language=job.target_language,
+        target_languages=job.target_languages,
         provider_id=job.provider_id,
         batch_size=job.batch_size or 50,
         upsert_strategy=job.upsert_strategy or "MERGE",
@@ -546,5 +577,462 @@ def get_datasource_columns(
     )
 # #endregion DatasourceColumnsService
 
+
+# #region InlineCorrectionService [C:3] [TYPE Class] [SEMANTICS translate, correction, inline, dictionary]
+# @BRIEF Service for inline editing translated values and submitting corrections to dictionaries.
+class InlineCorrectionService:
+    """Handle inline correction of translated values with optional dictionary submission."""
+
+    @staticmethod
+    def apply_inline_edit(
+        db: Session,
+        run_id: str,
+        record_id: str,
+        language_code: str,
+        final_value: str,
+        submit_to_dictionary: bool = False,
+        dictionary_id: str | None = None,
+        current_user: str | None = None,
+        context_data_override: dict | None = None,
+        usage_notes: str | None = None,
+        keep_context: bool = True,
+    ) -> dict:
+        """Apply an inline edit to a TranslationLanguage entry.
+
+        Updates final_value and user_edit fields. Optionally submits to dictionary.
+        Returns the updated TranslationLanguage data as a dict.
+
+        Args:
+            context_data_override: If provided, overrides auto-captured context data.
+            usage_notes: Usage notes for the dictionary entry.
+            keep_context: If False, clear context on the dictionary entry.
+        """
+        from ...models.translate import TranslationLanguage, TranslationRecord
+
+        # Find the language entry
+        lang_entry = (
+            db.query(TranslationLanguage)
+            .filter(
+                TranslationLanguage.record_id == record_id,
+                TranslationLanguage.language_code == language_code,
+            )
+            .first()
+        )
+        if not lang_entry:
+            raise ValueError(f"Language entry '{language_code}' not found for record '{record_id}'")
+
+        # Verify the record belongs to the run
+        record = (
+            db.query(TranslationRecord)
+            .filter(
+                TranslationRecord.id == record_id,
+                TranslationRecord.run_id == run_id,
+            )
+            .first()
+        )
+        if not record:
+            raise ValueError(
+                f"Translation record '{record_id}' not found in run '{run_id}'"
+            )
+
+        # Apply the edit
+        lang_entry.final_value = final_value
+        lang_entry.user_edit = final_value
+        if lang_entry.status in ("pending", "translated"):
+            lang_entry.status = "edited"
+        db.flush()
+
+        # Optionally submit to dictionary
+        dict_result = None
+        if submit_to_dictionary and dictionary_id:
+            try:
+                dict_result = InlineCorrectionService.submit_correction_to_dict(
+                    db=db,
+                    record_id=record_id,
+                    language_code=language_code,
+                    dictionary_id=dictionary_id,
+                    corrected_value=final_value,
+                    current_user=current_user,
+                    context_data_override=context_data_override,
+                    usage_notes=usage_notes,
+                    keep_context=keep_context,
+                )
+            except Exception as e:
+                # Dictionary submission failure should not block the edit
+                dict_result = {"error": str(e), "action": "failed"}
+
+        db.commit()
+        db.refresh(lang_entry)
+
+        from ...schemas.translate import TranslationLanguageResponse
+        response = TranslationLanguageResponse.model_validate(lang_entry)
+        result = response.model_dump()
+        if dict_result:
+            result["dictionary_result"] = dict_result
+        return result
+
+    @staticmethod
+    def submit_correction_to_dict(
+        db: Session,
+        record_id: str,
+        language_code: str,
+        dictionary_id: str,
+        corrected_value: str,
+        current_user: str | None = None,
+        context_data_override: dict | None = None,
+        usage_notes: str | None = None,
+        keep_context: bool = True,
+    ) -> dict:
+        """Submit a correction from an inline edit to the dictionary.
+
+        Fetches the TranslationLanguage entry, auto-captures source text and context,
+        and creates/updates a DictionaryEntry with correct language pair.
+        Returns conflict info if an entry already exists.
+
+        Args:
+            context_data_override: If provided, overrides auto-captured context_data.
+            usage_notes: Optional usage notes to store on the dictionary entry.
+            keep_context: If False, sets has_context=False and clears context_data.
+        """
+        from ...core.logger import belief_scope, logger
+        from ...models.translate import DictionaryEntry, TranslationLanguage, TranslationRecord
+        from ._utils import _normalize_term
+
+        with belief_scope("InlineCorrectionService.submit_correction_to_dict"):
+            # Find the language entry
+            lang_entry = (
+                db.query(TranslationLanguage)
+                .filter(
+                    TranslationLanguage.record_id == record_id,
+                    TranslationLanguage.language_code == language_code,
+                )
+                .first()
+            )
+            if not lang_entry:
+                raise ValueError(f"Language entry '{language_code}' not found for record '{record_id}'")
+
+            # Find the record for source text and context
+            record = (
+                db.query(TranslationRecord)
+                .filter(TranslationRecord.id == record_id)
+                .first()
+            )
+            if not record:
+                raise ValueError(f"Translation record '{record_id}' not found")
+
+            # Get source text from the record
+            source_term = record.source_sql or record.source_object_name or ""
+            if not source_term:
+                raise ValueError("No source text available for this record")
+
+            # Determine language pair
+            source_language = lang_entry.source_language_detected or "und"
+            target_language = language_code
+
+            # Prepare context data — auto-capture from source row
+            context_data = {}
+            if record.source_data:
+                context_data["source_data"] = record.source_data
+            if record.source_object_type:
+                context_data["source_object_type"] = record.source_object_type
+            if record.source_object_name:
+                context_data["source_object_name"] = record.source_object_name
+            if record.source_object_id:
+                context_data["source_object_id"] = record.source_object_id
+
+            # Apply context_data_override if provided (user edited)
+            if context_data_override is not None:
+                context_data = context_data_override
+                context_source = "manual"
+            else:
+                context_source = "auto"
+
+            # Handle keep_context=False (user explicitly removed context)
+            if not keep_context:
+                context_data = None
+                context_source = "manual"
+
+            # Normalize source term
+            normalized = _normalize_term(source_term)
+
+            # Check for existing entry with same (dictionary_id, source_term_norm, source_language, target_language)
+            existing = (
+                db.query(DictionaryEntry)
+                .filter(
+                    DictionaryEntry.dictionary_id == dictionary_id,
+                    DictionaryEntry.source_term_normalized == normalized,
+                    DictionaryEntry.source_language == source_language,
+                    DictionaryEntry.target_language == target_language,
+                )
+                .first()
+            )
+
+            result = {
+                "action": "created",
+                "entry_id": None,
+                "conflict": None,
+                "message": None,
+            }
+
+            if existing:
+                # Conflict: return conflict info
+                result["action"] = "conflict_detected"
+                result["conflict"] = {
+                    "source_term": source_term,
+                    "existing_target_term": existing.target_term,
+                    "submitted_target_term": corrected_value,
+                    "action": "keep_existing",
+                }
+                result["message"] = (
+                    f"Existing entry found: '{existing.target_term}' "
+                    f"for source '{source_term}' ({source_language} → {target_language})"
+                )
+                logger.reason("Correction conflict detected", result)
+                return result
+
+            # Create new entry
+            entry = DictionaryEntry(
+                dictionary_id=dictionary_id,
+                source_term=source_term.strip(),
+                source_term_normalized=normalized,
+                target_term=corrected_value.strip(),
+                source_language=source_language,
+                target_language=target_language,
+                context_data=context_data if context_data else None,
+                context_notes=None,
+                has_context=bool(context_data) and keep_context,
+                context_source=context_source,
+                usage_notes=usage_notes,
+                origin_source_language=source_language,
+                origin_run_id=record.run_id,
+                origin_row_key=record.id,
+                origin_user_id=current_user,
+            )
+            db.add(entry)
+            db.flush()
+            result["entry_id"] = entry.id
+            result["message"] = (
+                f"Entry created for '{source_term}' ({source_language}) → "
+                f"'{corrected_value}' ({target_language})"
+            )
+            logger.reflect("Dictionary entry created from correction", result)
+            return result
+# #endregion InlineCorrectionService
+
+
+# #region BulkFindReplaceService [C:3] [TYPE Class] [SEMANTICS translate, bulk, find, replace, regex]
+# @BRIEF Service for bulk find-and-replace operations on translated values.
+class BulkFindReplaceService:
+    """Handle bulk find-and-replace on TranslationLanguage entries."""
+
+    @staticmethod
+    def _compile_pattern(pattern: str, is_regex: bool) -> re.Pattern:
+        """Compile a search pattern (regex or plain text)."""
+        import re
+        if is_regex:
+            return re.compile(pattern)
+        return re.compile(re.escape(pattern))
+
+    @staticmethod
+    def _find_matching_entries(
+        db: Session,
+        run_id: str,
+        pattern: str,
+        is_regex: bool,
+        target_language: str,
+    ) -> list[Any]:
+        """Find all TranslationLanguage entries matching the pattern."""
+        from ...models.translate import TranslationLanguage, TranslationRecord
+
+        # We scan entries for the target language in this run
+        entries = (
+            db.query(TranslationLanguage)
+            .join(
+                TranslationRecord,
+                TranslationRecord.id == TranslationLanguage.record_id,
+            )
+            .filter(
+                TranslationRecord.run_id == run_id,
+                TranslationLanguage.language_code == target_language,
+            )
+            .all()
+        )
+        return entries
+
+    @staticmethod
+    def preview(
+        db: Session,
+        run_id: str,
+        pattern: str,
+        is_regex: bool,
+        target_language: str,
+    ) -> list[dict]:
+        """Scan TranslationLanguage entries and return matching items without applying changes."""
+        from ...models.translate import TranslationRecord
+
+        compiled = BulkFindReplaceService._compile_pattern(pattern, is_regex)
+        entries = BulkFindReplaceService._find_matching_entries(
+            db, run_id, pattern, is_regex, target_language
+        )
+
+        preview_items = []
+        for entry in entries:
+            current_value = entry.final_value or entry.translated_value or ""
+            if compiled.search(current_value):
+                # Get source term for context
+                record = (
+                    db.query(TranslationRecord)
+                    .filter(TranslationRecord.id == entry.record_id)
+                    .first()
+                )
+                source_term = record.source_sql if record else ""
+
+                new_value = compiled.sub(
+                    BulkFindReplaceService._get_replacement(pattern, is_regex),
+                    current_value,
+                )
+                if new_value != current_value:
+                    preview_items.append({
+                        "record_id": entry.record_id,
+                        "language_code": entry.language_code,
+                        "source_term": source_term or "",
+                        "current_value": current_value,
+                        "new_value": new_value,
+                        "source_language_detected": entry.source_language_detected,
+                    })
+
+        return preview_items
+
+    @staticmethod
+    def _get_replacement(pattern: str, is_regex: bool) -> str:
+        """Return replacement pattern (for regex, use backrefs; for plain, use literal)."""
+        # The replacement_text is passed through at apply time
+        return pattern  # Placeholder — actual replacement is done in apply()
+
+    @staticmethod
+    def apply(
+        db: Session,
+        run_id: str,
+        pattern: str,
+        is_regex: bool,
+        replacement_text: str,
+        target_language: str,
+        submit_to_dictionary: bool = False,
+        dictionary_id: str | None = None,
+        usage_notes: str | None = None,
+        current_user: str | None = None,
+        submit_to_dictionary_with_context: bool = False,
+    ) -> dict:
+        """Apply find-and-replace to matching TranslationLanguage entries.
+
+        Returns dict with rows_affected, corrections_submitted, and preview list.
+
+        Args:
+            submit_to_dictionary_with_context: If True, auto-capture source row context
+                when submitting to dictionary.
+        """
+        from ...core.logger import belief_scope, logger
+        from ...models.translate import DictionaryEntry, TranslationRecord
+
+        with belief_scope("BulkFindReplaceService.apply"):
+            compiled = BulkFindReplaceService._compile_pattern(pattern, is_regex)
+            entries = BulkFindReplaceService._find_matching_entries(
+                db, run_id, pattern, is_regex, target_language
+            )
+
+            preview_items = []
+            rows_affected = 0
+            corrections_submitted = 0
+
+            # Track unique (source_term, replacement) -> first record for context capture
+            seen_term_replacements: dict[str, str] = {}
+
+            for entry in entries:
+                current_value = entry.final_value or entry.translated_value or ""
+                if compiled.search(current_value):
+                    new_value = compiled.sub(replacement_text, current_value)
+                    if new_value != current_value:
+                        # Apply the replacement
+                        entry.final_value = new_value
+                        entry.user_edit = new_value
+                        if entry.status in ("pending", "translated"):
+                            entry.status = "edited"
+                        rows_affected += 1
+
+                        # Get source term for preview
+                        record = (
+                            db.query(TranslationRecord)
+                            .filter(TranslationRecord.id == entry.record_id)
+                            .first()
+                        )
+                        source_term = record.source_sql if record else ""
+
+                        # Optionally submit to dictionary
+                        if submit_to_dictionary and dictionary_id:
+                            try:
+                                dict_result = InlineCorrectionService.submit_correction_to_dict(
+                                    db=db,
+                                    record_id=entry.record_id,
+                                    language_code=entry.language_code,
+                                    dictionary_id=dictionary_id,
+                                    corrected_value=new_value,
+                                    current_user=current_user,
+                                )
+                                if dict_result.get("action") in ("created", "updated"):
+                                    corrections_submitted += 1
+
+                                # Track context for unique (source_term, replacement) pairs
+                                if submit_to_dictionary_with_context and record:
+                                    term_key = f"{source_term}::{new_value}"
+                                    if term_key not in seen_term_replacements:
+                                        seen_term_replacements[term_key] = entry.record_id
+                                        # Auto-capture context from this row
+                                        context_data = {}
+                                        if record.source_data:
+                                            context_data["source_data"] = record.source_data
+                                        if record.source_object_type:
+                                            context_data["source_object_type"] = record.source_object_type
+                                        if record.source_object_name:
+                                            context_data["source_object_name"] = record.source_object_name
+                                        if record.source_object_id:
+                                            context_data["source_object_id"] = record.source_object_id
+                                        # Update the just-created entry's context
+                                        if dict_result.get("action") == "created" and dict_result.get("entry_id"):
+                                            d_entry = (
+                                                db.query(DictionaryEntry)
+                                                .filter(DictionaryEntry.id == dict_result["entry_id"])
+                                                .first()
+                                            )
+                                            if d_entry:
+                                                d_entry.context_data = context_data if context_data else None
+                                                d_entry.has_context = bool(context_data)
+                                                d_entry.context_source = "bulk"
+                            except Exception as e:
+                                logger.explore(
+                                    "Bulk replace: dictionary submission failed",
+                                    extra={"record_id": entry.record_id, "error": str(e)},
+                                )
+
+                        preview_items.append({
+                            "record_id": entry.record_id,
+                            "language_code": entry.language_code,
+                            "source_term": source_term or "",
+                            "current_value": current_value,
+                            "new_value": new_value,
+                            "source_language_detected": entry.source_language_detected,
+                        })
+
+            db.commit()
+
+            result = {
+                "rows_affected": rows_affected,
+                "corrections_submitted": corrections_submitted,
+                "preview": preview_items,
+            }
+            logger.reason("Bulk replace completed", result)
+            return result
+# #endregion BulkFindReplaceService
+
+
 # #endregion TranslateJobService
 # #endregion TranslateJobService