feat(translate,scheduler,trace): new-key-only mode, stale PENDING protection, trace_id propagation

Translations:
- FR-045: new-key-only execution mode — translate only rows with unseen keys
- Compare source row keys against TranslationRecord.source_data from last succeeded run
- Baseline expired (>90 days) fallback to full mode with baseline_expired event
- run_noop early return when zero new keys (skip LLM + SQL)

Scheduler reliability:
- Stale PENDING protection: runs older than 1h auto-marked FAILED, no longer block schedule
- load_schedules() reloads active translation schedules from DB on restart
- add_translation_job/remove_translation_job register/unregister with APScheduler
- execution_mode column on translation_schedules with additive DB migration

Automation view:
- GET /settings/automation/translation-schedules endpoint
- Translation schedules displayed on Automation page below validation policies

Trace propagation:
- seed_trace_id() in all background entry points:
  TaskManager._run_task/_flusher_loop, SchedulerService._trigger_backup,
  websocket_endpoint, IdMappingService, all standalone scripts

Migrations:
- dictionary_entries: origin_run_id, origin_row_key, origin_user_id
- translation_schedules: execution_mode

Tests:
- 3 new test modules (22 tests): core_scheduler, executor_filter, scheduler_execution+guard
- Fix pre-existing translate test isolation (conftest.py)
- Fix test_list_runs_filter_status parameter

backend/src/plugins/translate/executor.py

@@ -40,6 +40,7 @@ from .preview import DEFAULT_EXECUTION_PROMPT_TEMPLATE
 # #region MAX_RETRIES_PER_BATCH [TYPE Constant]
 # @BRIEF Maximum number of retries for a single batch before marking it failed.
 MAX_RETRIES_PER_BATCH = 3
+# #endregion MAX_RETRIES_PER_BATCH
 
 
 # #region TranslationExecutor [C:4] [TYPE Class]
@@ -97,6 +98,20 @@ class TranslationExecutor:
                 self.db.flush()
                 return run
 
+            # Apply new-key-only filtering for scheduled runs
+            if run.trigger_type == "new_key_only":
+                source_rows = self._filter_new_keys(job, run.id, source_rows)
+                if not source_rows:
+                    logger.reason("run_noop — no new rows to translate", {
+                        "job_id": job.id, "run_id": run.id,
+                    })
+                    run.status = "COMPLETED"
+                    run.insert_status = "skipped"
+                    run.total_records = 0
+                    run.completed_at = datetime.now(timezone.utc)
+                    self.db.commit()
+                    return
+
             total_rows = len(source_rows)
             run.total_records = total_rows
 
@@ -303,6 +318,74 @@ class TranslationExecutor:
             return source_rows
     # endregion _fetch_source_rows
 
+    # region _filter_new_keys [TYPE Function]
+    # @PURPOSE: Filter source rows to only include those with keys absent from the last successful run.
+    # @PRE: job and run are persisted; source_rows is a list of dicts with source_data containing key columns.
+    # @POST: Returns filtered list of source rows with only new keys. Logs skip count.
+    # @SIDE_EFFECT: Queries TranslationRecord from previous runs; no writes.
+    def _filter_new_keys(self, job, run_id: str, source_rows: list) -> list:
+        with belief_scope("TranslationExecutor._filter_new_keys"):
+            # Find most recent COMPLETED run with successful insert for this job
+            prev_run = (
+                self.db.query(TranslationRun)
+                .filter(
+                    TranslationRun.job_id == job.id,
+                    TranslationRun.status == "COMPLETED",
+                    TranslationRun.insert_status == "succeeded",
+                    TranslationRun.id != run_id,
+                )
+                .order_by(TranslationRun.created_at.desc())
+                .first()
+            )
+            if not prev_run:
+                logger.reason("No prior successful run — all keys treated as new", {
+                    "job_id": job.id,
+                })
+                return source_rows
+
+            # Get successful records from that run
+            prev_records = (
+                self.db.query(TranslationRecord)
+                .filter(
+                    TranslationRecord.run_id == prev_run.id,
+                    TranslationRecord.status == "SUCCESS",
+                )
+                .all()
+            )
+            if not prev_records:
+                return source_rows
+
+            # Build set of already-translated composite keys
+            key_cols = job.target_key_cols or job.source_key_cols or []
+            if not key_cols:
+                logger.explore("No key columns configured — skipping new-key-only filter",
+                               {"job_id": job.id})
+                return source_rows
+            existing_keys = set()
+            for rec in prev_records:
+                sd = rec.source_data or {}
+                key_tuple = tuple(str(sd.get(k, "")) for k in key_cols)
+                existing_keys.add(key_tuple)
+
+            # Filter: keep only rows whose keys are NOT in the existing set
+            filtered = []
+            skipped = 0
+            for row in source_rows:
+                sd = row.get("source_data", {}) or {}
+                key_tuple = tuple(str(sd.get(k, "")) for k in key_cols)
+                if key_tuple not in existing_keys:
+                    filtered.append(row)
+                else:
+                    skipped += 1
+
+            logger.reason(f"New-key-only filter: {len(source_rows)} total → {len(filtered)} new, {skipped} skipped", {
+                "job_id": job.id,
+                "prev_run_id": prev_run.id,
+                "key_cols": key_cols,
+            })
+            return filtered
+    # endregion _filter_new_keys
+
     # region _extract_chart_data_rows [TYPE Function]
     # @PURPOSE: Extract data rows from Superset chart data API response.
     # @POST: Returns list of dicts with column-value pairs.
@@ -763,4 +846,3 @@ class TranslationExecutor:
 
 # #endregion TranslationExecutor
 # #endregion TranslationExecutor
-# #endregion TranslationExecutor