diff --git a/backend/src/plugins/translate/_batch_insert.py b/backend/src/plugins/translate/_batch_insert.py
index 96fa043a..35b6fa3f 100644
--- a/backend/src/plugins/translate/_batch_insert.py
+++ b/backend/src/plugins/translate/_batch_insert.py
@@ -49,11 +49,25 @@ def insert_batch_to_target(
         if dialect is None:
             return
 
-        sql, row_count = _generate_insert_sql(dialect, job, columns, rows_for_sql, batch_id)
-        if sql is None:
-            return
-
-        _execute_insert_sql(executor, sql, batch_id, row_count)
+        # Use chunked INSERT to avoid SQL size limits (max 500 VALUES rows per statement)
+        statements = SQLGenerator.generate_batch(
+            dialect=dialect,
+            target_schema=job.target_schema,
+            target_table=job.target_table or "translated_data",
+            columns=columns,
+            rows=rows_for_sql,
+            key_columns=job.target_key_cols,
+            upsert_strategy=job.upsert_strategy or "MERGE",
+            max_rows_per_statement=500,
+        )
+        total_inserted = 0
+        for sql, chunk_count in statements:
+            if sql is None:
+                continue
+            _execute_insert_sql(executor, sql, batch_id, chunk_count)
+            total_inserted += chunk_count
+        logger.reason(f"Batch {batch_id[:12]} inserted {total_inserted} rows",
+                      {"batch_id": batch_id, "rows": total_inserted, "chunks": len(statements)})
 # #endregion insert_batch_to_target
 
 
diff --git a/backend/src/plugins/translate/orchestrator_config.py b/backend/src/plugins/translate/orchestrator_config.py
index 47f60fa1..01e81bc8 100644
--- a/backend/src/plugins/translate/orchestrator_config.py
+++ b/backend/src/plugins/translate/orchestrator_config.py
@@ -30,18 +30,40 @@ def compute_config_hash(job) -> str:
 
 # #region compute_dict_snapshot_hash [C:2] [TYPE Function] [SEMANTICS dictionary, hash, snapshot]
 # @BRIEF Compute a hash of dictionary state for snapshot comparison.
+#         Includes dictionary IDs, entry count, and max entry updated_at
+#         to invalidate cache when dictionary entries are modified.
 # @PRE db_session is valid.
-# @POST Returns 16-char hex SHA-256 hash of concatenated dictionary IDs.
+# @POST Returns 16-char hex SHA-256 hash.
 def compute_dict_snapshot_hash(db_session, job_id: str) -> str:
     """Compute a hash of dictionary state for snapshot comparison."""
-    from ...models.translate import TranslationJobDictionary
+    from ...models.translate import TranslationJobDictionary, DictionaryEntry
+    from sqlalchemy import func
+
     dict_links = (
         db_session.query(TranslationJobDictionary)
         .filter(TranslationJobDictionary.job_id == job_id)
         .all()
     )
-    dict_ids = sorted([dl.dictionary_id for dl in dict_links])
-    hash_input = ",".join(dict_ids)
+    if not dict_links:
+        return hashlib.sha256(b"").hexdigest()[:16]
+
+    parts: list[str] = []
+    for dl in sorted(dict_links, key=lambda d: d.dictionary_id or ""):
+        # Include entry count + max updated_at to detect content changes
+        entry_count = (
+            db_session.query(func.count(DictionaryEntry.id))
+            .filter(DictionaryEntry.dictionary_id == dl.dictionary_id)
+            .scalar()
+        )
+        max_updated = (
+            db_session.query(func.max(DictionaryEntry.updated_at))
+            .filter(DictionaryEntry.dictionary_id == dl.dictionary_id)
+            .scalar()
+        )
+        max_updated_str = max_updated.isoformat() if max_updated else "0"
+        parts.append(f"{dl.dictionary_id}:c{entry_count}:u{max_updated_str}")
+
+    hash_input = "|".join(parts)
     return hashlib.sha256(hash_input.encode()).hexdigest()[:16]
 # #endregion compute_dict_snapshot_hash
 # #endregion orchestrator_config
diff --git a/backend/src/plugins/translate/preview_response_parser.py b/backend/src/plugins/translate/preview_response_parser.py
index 0263aa57..8292d96f 100644
--- a/backend/src/plugins/translate/preview_response_parser.py
+++ b/backend/src/plugins/translate/preview_response_parser.py
@@ -123,15 +123,36 @@ def compute_config_hash(job: TranslationJob) -> str:
 
 # #region compute_dict_snapshot_hash [C:2] [TYPE Function] [SEMANTICS dictionary, hash, snapshot]
 # @BRIEF Compute a hash of dictionary state for snapshot comparison.
+#         Includes dictionary IDs, entry count, and max entry updated_at.
 def compute_dict_snapshot_hash(db: Session, job_id: str) -> str:
     """Compute a hash of dictionary state for snapshot comparison."""
+    from sqlalchemy import func
+    from ...models.translate import TranslationJobDictionary, DictionaryEntry
+
     dict_links = (
         db.query(TranslationJobDictionary)
         .filter(TranslationJobDictionary.job_id == job_id)
         .all()
     )
-    dict_ids = sorted([dl.dictionary_id for dl in dict_links])
-    hash_input = ",".join(dict_ids)
+    if not dict_links:
+        return hashlib.sha256(b"").hexdigest()[:16]
+
+    parts: list[str] = []
+    for dl in sorted(dict_links, key=lambda d: d.dictionary_id or ""):
+        entry_count = (
+            db.query(func.count(DictionaryEntry.id))
+            .filter(DictionaryEntry.dictionary_id == dl.dictionary_id)
+            .scalar()
+        )
+        max_updated = (
+            db.query(func.max(DictionaryEntry.updated_at))
+            .filter(DictionaryEntry.dictionary_id == dl.dictionary_id)
+            .scalar()
+        )
+        max_updated_str = max_updated.isoformat() if max_updated else "0"
+        parts.append(f"{dl.dictionary_id}:c{entry_count}:u{max_updated_str}")
+
+    hash_input = "|".join(parts)
     return hashlib.sha256(hash_input.encode()).hexdigest()[:16]
 # #endregion compute_dict_snapshot_hash
 # #endregion preview_response_parser