diff --git a/backend/src/api/routes/translate/_schedule_routes.py b/backend/src/api/routes/translate/_schedule_routes.py index a6a6eba1..7204d1ad 100644 --- a/backend/src/api/routes/translate/_schedule_routes.py +++ b/backend/src/api/routes/translate/_schedule_routes.py @@ -44,6 +44,7 @@ async def get_schedule( "cron_expression": schedule.cron_expression, "timezone": schedule.timezone, "is_active": schedule.is_active, + "execution_mode": schedule.execution_mode, "last_run_at": schedule.last_run_at.isoformat() if schedule.last_run_at else None, "next_run_at": schedule.next_run_at.isoformat() if schedule.next_run_at else None, "created_by": schedule.created_by, @@ -77,12 +78,14 @@ async def set_schedule( existing = db.query(TranslationSchedule).filter( TranslationSchedule.job_id == job_id ).first() + execution_mode = getattr(payload, 'execution_mode', 'full') if existing: schedule = scheduler.update_schedule( job_id, cron_expression=payload.cron_expression, timezone_str=payload.timezone, is_active=payload.is_active, + execution_mode=execution_mode, ) else: schedule = scheduler.create_schedule( @@ -90,6 +93,7 @@ async def set_schedule( cron_expression=payload.cron_expression, timezone=payload.timezone, is_active=payload.is_active, + execution_mode=execution_mode, ) # Register with APScheduler via SchedulerService from ....dependencies import get_scheduler_service @@ -99,6 +103,7 @@ async def set_schedule( job_id=job_id, cron_expression=schedule.cron_expression, timezone=schedule.timezone, + execution_mode=schedule.execution_mode, ) return { "id": schedule.id, @@ -106,6 +111,7 @@ async def set_schedule( "cron_expression": schedule.cron_expression, "timezone": schedule.timezone, "is_active": schedule.is_active, + "execution_mode": schedule.execution_mode, "last_run_at": schedule.last_run_at.isoformat() if schedule.last_run_at else None, "created_by": schedule.created_by, "created_at": schedule.created_at.isoformat() if schedule.created_at else None, @@ -140,6 +146,7 @@ async def enable_schedule( job_id=job_id, cron_expression=schedule.cron_expression, timezone=schedule.timezone, + execution_mode=schedule.execution_mode, ) return {"status": "enabled", "job_id": job_id} except ValueError as e: diff --git a/backend/src/app.py b/backend/src/app.py index 7f380174..44c4baff 100755 --- a/backend/src/app.py +++ b/backend/src/app.py @@ -272,6 +272,7 @@ async def websocket_endpoint( source: Filter logs by source component (e.g., "plugin", "superset_api") level: Filter logs by minimum level (DEBUG, INFO, WARNING, ERROR) """ + seed_trace_id() with belief_scope("websocket_endpoint", f"task_id={task_id}"): await websocket.accept() source_filter = source.lower() if source else None diff --git a/backend/src/core/database.py b/backend/src/core/database.py index 4f4d7014..c3ae7bd3 100644 --- a/backend/src/core/database.py +++ b/backend/src/core/database.py @@ -579,6 +579,96 @@ def _ensure_dataset_review_session_columns(bind_engine): # #endregion _ensure_dataset_review_session_columns +# #region _ensure_translation_schedules_columns [C:3] [TYPE Function] +# @BRIEF Applies additive schema upgrades for translation_schedules table. +# @PRE: bind_engine points to application database. +# @POST: Missing columns are added without data loss. +# @RELATION DEPENDS_ON -> [engine] +def _ensure_translation_schedules_columns(bind_engine): + with belief_scope("_ensure_translation_schedules_columns"): + table_name = "translation_schedules" + inspector = inspect(bind_engine) + if table_name not in inspector.get_table_names(): + return + + existing_columns = { + str(column.get("name") or "").strip() + for column in inspector.get_columns(table_name) + } + + alter_statements = [] + if "execution_mode" not in existing_columns: + alter_statements.append( + "ALTER TABLE translation_schedules " + "ADD COLUMN execution_mode VARCHAR NOT NULL DEFAULT 'full'" + ) + + if not alter_statements: + return + + try: + with bind_engine.begin() as connection: + for statement in alter_statements: + connection.execute(text(statement)) + except Exception as migration_error: + logger.explore( + "TranslationSchedule additive migration failed", + extra={"src": "database", "error": str(migration_error)}, + ) + + +# #endregion _ensure_translation_schedules_columns + + +# #region _ensure_dictionary_entries_columns [C:3] [TYPE Function] +# @BRIEF Additive migration for dictionary_entries origin tracking columns. +# @RELATION DEPENDS_ON -> [engine] +def _ensure_dictionary_entries_columns(bind_engine): + with belief_scope("_ensure_dictionary_entries_columns"): + table_name = "dictionary_entries" + inspector = inspect(bind_engine) + if table_name not in inspector.get_table_names(): + return + + existing_columns = { + str(column.get("name") or "").strip() + for column in inspector.get_columns(table_name) + } + + alter_statements = [] + if "origin_run_id" not in existing_columns: + alter_statements.append( + "ALTER TABLE dictionary_entries " + "ADD COLUMN origin_run_id VARCHAR" + ) + if "origin_row_key" not in existing_columns: + alter_statements.append( + "ALTER TABLE dictionary_entries " + "ADD COLUMN origin_row_key VARCHAR" + ) + if "origin_user_id" not in existing_columns: + alter_statements.append( + "ALTER TABLE dictionary_entries " + "ADD COLUMN origin_user_id VARCHAR" + ) + + if not alter_statements: + return + + try: + with bind_engine.begin() as connection: + for statement in alter_statements: + connection.execute(text(statement)) + except Exception as migration_error: + logger.explore( + "DictionaryEntry additive migration failed", + extra={"src": "database", "error": str(migration_error)}, + ) + + +# #endregion _ensure_dictionary_entries_columns + + # #region init_db [C:3] [TYPE Function] # @BRIEF Initializes the database by creating all tables. # @PRE: engine, tasks_engine and auth_engine are initialized. @@ -587,6 +677,7 @@ def _ensure_dataset_review_session_columns(bind_engine): # @RELATION CALLS -> [ensure_connection_configs_table] # @RELATION CALLS -> [_ensure_filter_source_enum_values] # @RELATION CALLS -> [_ensure_dataset_review_session_columns] +# @RELATION CALLS -> [_ensure_dictionary_entries_columns] def init_db(): with belief_scope("init_db"): Base.metadata.create_all(bind=engine) @@ -601,6 +692,8 @@ def init_db(): _ensure_filter_source_enum_values(engine) _ensure_dataset_review_session_columns(engine) _ensure_translation_jobs_columns(engine) + _ensure_translation_schedules_columns(engine) + _ensure_dictionary_entries_columns(engine) # #endregion init_db diff --git a/backend/src/core/mapping_service.py b/backend/src/core/mapping_service.py index 62c57797..2d18481c 100644 --- a/backend/src/core/mapping_service.py +++ b/backend/src/core/mapping_service.py @@ -18,6 +18,7 @@ from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from src.models.mapping import ResourceMapping, ResourceType from src.core.logger import logger, belief_scope +from src.core.cot_logger import seed_trace_id # #region IdMappingService [C:5] [TYPE Class] # @BRIEF Service handling the cataloging and retrieval of remote Superset Integer IDs. # @PRE: db_session is an active SQLAlchemy Session bound to mapping tables. @@ -65,6 +66,7 @@ class IdMappingService: "[IdMappingService.start_scheduler][Reflect] Removed existing sync job." ) def sync_all(): + seed_trace_id() for env_id in environments: client = superset_client_factory(env_id) if client: @@ -98,6 +100,7 @@ class IdMappingService: Polls the Superset APIs for the target environment and updates the local mapping table. If incremental=True, only fetches items changed since the max last_synced_at date. """ + seed_trace_id() with belief_scope("IdMappingService.sync_environment"): logger.reason( f"Starting sync for environment {environment_id}", diff --git a/backend/src/core/scheduler.py b/backend/src/core/scheduler.py index a6976f95..c8c506fc 100644 --- a/backend/src/core/scheduler.py +++ b/backend/src/core/scheduler.py @@ -5,6 +5,7 @@ from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from .logger import logger, belief_scope +from .cot_logger import seed_trace_id from .config_manager import ConfigManager from .database import SessionLocal import asyncio @@ -45,10 +46,11 @@ class SchedulerService: self.scheduler.shutdown() logger.info("Scheduler stopped.") # #endregion stop - # #region load_schedules [TYPE Function] - # @PURPOSE: Loads backup schedules from configuration and registers them. - # @PRE: config_manager must have valid configuration. - # @POST: All enabled backup jobs are added to the scheduler. + # #region load_schedules [C:4] [TYPE Function] [SEMANTICS scheduler,backup,translation,apscheduler] + # @BRIEF Load backup and active translation schedules from config and DB, re-registering all jobs. + # @PRE config_manager must have valid configuration; database is accessible. + # @POST All enabled backup jobs and active translation schedules are re-registered in APScheduler. + # @SIDE_EFFECT Removes all existing APScheduler jobs; queries translation_schedules table; registers APScheduler jobs for translation. def load_schedules(self): with belief_scope("SchedulerService.load_schedules"): # Clear existing jobs @@ -57,6 +59,31 @@ class SchedulerService: for env in config.environments: if env.backup_schedule and env.backup_schedule.enabled: self.add_backup_job(env.id, env.backup_schedule.cron_expression) + # Re-register active translation schedules from the database + try: + from ..models.translate import TranslationSchedule + db = SessionLocal() + try: + active_schedules = ( + db.query(TranslationSchedule) + .filter(TranslationSchedule.is_active == True) + .all() + ) + for sched in active_schedules: + self.add_translation_job( + schedule_id=sched.id, + job_id=sched.job_id, + cron_expression=sched.cron_expression, + timezone=sched.timezone, + execution_mode=getattr(sched, 'execution_mode', 'full'), + ) + if active_schedules: + logger.reason(f"Loaded {len(active_schedules)} active translation schedule(s)") + finally: + db.close() + except Exception as e: + logger.explore("Failed to load translation schedules on startup", + extra={"error": str(e)}) # #endregion load_schedules # #region add_backup_job [TYPE Function] # @PURPOSE: Adds a scheduled backup job for an environment. @@ -91,7 +118,7 @@ class SchedulerService: # @SIDE_EFFECT Mutates APScheduler state; calls execute_scheduled_translation on trigger. # @RELATION DEPENDS_ON -> [execute_scheduled_translation] # @RELATION DEPENDS_ON -> [SessionLocal] - def add_translation_job(self, schedule_id: str, job_id: str, cron_expression: str, timezone: str = "UTC"): + def add_translation_job(self, schedule_id: str, job_id: str, cron_expression: str, timezone: str = "UTC", execution_mode: str = "full"): with belief_scope( "SchedulerService.add_translation_job", f"schedule_id={schedule_id}, job_id={job_id}, cron={cron_expression}", @@ -106,7 +133,7 @@ class SchedulerService: execute_scheduled_translation, CronTrigger.from_crontab(cron_expression, timezone=tz), id=job_id_aps, - args=[schedule_id, job_id, SessionLocal, self.config_manager], + args=[schedule_id, job_id, SessionLocal, self.config_manager, execution_mode], replace_existing=True, ) logger.reason( @@ -141,6 +168,7 @@ class SchedulerService: # @POST: A new backup task is created in the task manager if not already running. # @PARAM: env_id (str) - The ID of the environment. def _trigger_backup(self, env_id: str): + seed_trace_id() with belief_scope("SchedulerService._trigger_backup", f"env_id={env_id}"): logger.info(f"Triggering scheduled backup for environment {env_id}") # Check if a backup is already running for this environment diff --git a/backend/src/core/task_manager/manager.py b/backend/src/core/task_manager/manager.py index d4f5b3c1..7c30a5ea 100644 --- a/backend/src/core/task_manager/manager.py +++ b/backend/src/core/task_manager/manager.py @@ -34,6 +34,7 @@ from .models import Task, TaskStatus, LogEntry, LogFilter, LogStats from .persistence import TaskPersistenceService, TaskLogPersistenceService from .context import TaskContext from ..logger import logger, belief_scope, should_log_task_level +from ..cot_logger import seed_trace_id # #region TaskManager [C:5] [TYPE Class] [SEMANTICS task, manager, lifecycle, execution, state] # @BRIEF Manages the lifecycle of tasks, including their creation, execution, and state tracking. # @LAYER: Core @@ -138,6 +139,7 @@ class TaskManager: # @RELATION: [DEPENDS_ON] ->[EventBus] def _flusher_loop(self): """Background thread that flushes log buffer to database.""" + seed_trace_id() # seed trace_id for the daemon flusher thread while not self._flusher_stop_event.is_set(): self._flush_logs() self._flusher_stop_event.wait(self.LOG_FLUSH_INTERVAL) @@ -150,6 +152,7 @@ class TaskManager: # @RELATION: [DEPENDS_ON] ->[EventBus] def _flush_logs(self): """Flush all buffered logs to the database.""" + seed_trace_id() with self._log_buffer_lock: task_ids = list(self._log_buffer.keys()) for task_id in task_ids: @@ -232,6 +235,7 @@ class TaskManager: # @RELATION: [DEPENDS_ON] ->[TaskContext] # @RELATION: [DEPENDS_ON] ->[EventBus] async def _run_task(self, task_id: str): + seed_trace_id() with belief_scope("TaskManager._run_task", f"task_id={task_id}"): task = self.tasks[task_id] plugin = self.plugin_loader.get_plugin(task.plugin_id) diff --git a/backend/src/models/translate.py b/backend/src/models/translate.py index 25c20232..9c8c3b31 100644 --- a/backend/src/models/translate.py +++ b/backend/src/models/translate.py @@ -239,6 +239,7 @@ class TranslationSchedule(Base): is_active = Column(Boolean, default=True) last_run_at = Column(DateTime, nullable=True) next_run_at = Column(DateTime, nullable=True) + execution_mode = Column(String, nullable=False, default="full", comment="full, new_key_only") created_by = Column(String, nullable=True) created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc)) updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)) diff --git a/backend/src/plugins/translate/executor.py b/backend/src/plugins/translate/executor.py index 0901dfa9..d1aeddd0 100644 --- a/backend/src/plugins/translate/executor.py +++ b/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 diff --git a/backend/src/plugins/translate/scheduler.py b/backend/src/plugins/translate/scheduler.py index 9b97a31a..e60a8e00 100644 --- a/backend/src/plugins/translate/scheduler.py +++ b/backend/src/plugins/translate/scheduler.py @@ -19,6 +19,7 @@ from typing import Any, Dict, List, Optional from sqlalchemy.orm import Session from ...core.logger import logger, belief_scope +from ...core.cot_logger import seed_trace_id from ...core.config_manager import ConfigManager from ...models.translate import TranslationSchedule, TranslationJob, TranslationRun from .events import TranslationEventLog @@ -44,6 +45,7 @@ class TranslationScheduler: cron_expression: str, timezone: str = "UTC", is_active: bool = True, + execution_mode: str = "full", ) -> TranslationSchedule: with belief_scope("TranslationScheduler.create_schedule"): # Verify job exists @@ -57,6 +59,7 @@ class TranslationScheduler: cron_expression=cron_expression, timezone=timezone, is_active=is_active, + execution_mode=execution_mode, created_by=self.current_user, ) self.db.add(schedule) @@ -88,6 +91,7 @@ class TranslationScheduler: cron_expression: Optional[str] = None, timezone_str: Optional[str] = None, is_active: Optional[bool] = None, + execution_mode: Optional[str] = None, ) -> TranslationSchedule: with belief_scope("TranslationScheduler.update_schedule"): schedule = self.db.query(TranslationSchedule).filter( @@ -102,6 +106,8 @@ class TranslationScheduler: schedule.timezone = timezone_str if is_active is not None: schedule.is_active = is_active + if execution_mode is not None: + schedule.execution_mode = execution_mode schedule.updated_at = datetime.now(timezone.utc) self.db.commit() self.db.refresh(schedule) @@ -245,8 +251,10 @@ def execute_scheduled_translation( job_id: str, db_session_maker, config_manager: ConfigManager, + execution_mode: str = "full", ) -> None: """APScheduler job callback for scheduled translations.""" + seed_trace_id() db: Session = db_session_maker() try: with belief_scope("execute_scheduled_translation"): @@ -265,26 +273,57 @@ def execute_scheduled_translation( }) # Concurrency check: max 1 pending/running run per job + STALE_THRESHOLD = timedelta(hours=1) + now = datetime.now(timezone.utc) + active_run = ( db.query(TranslationRun) .filter( TranslationRun.job_id == job_id, TranslationRun.status.in_(["PENDING", "RUNNING"]), ) + .order_by(TranslationRun.created_at.asc()) # oldest first .first() ) if active_run: - logger.explore("Skipping scheduled run — concurrent run in progress", { - "job_id": job_id, - "existing_run_id": active_run.id, - }) - event_log = TranslationEventLog(db) - event_log.log_event( - job_id=job_id, - event_type="RUN_STARTED", - payload={"reason": "skipped_concurrent", "existing_run_id": active_run.id}, + is_stale = ( + active_run.status == "PENDING" + and active_run.created_at + and (now - active_run.created_at) > STALE_THRESHOLD ) - return + if is_stale: + # Mark ALL stale PENDING runs as FAILED + stale_runs = ( + db.query(TranslationRun) + .filter( + TranslationRun.job_id == job_id, + TranslationRun.status == "PENDING", + TranslationRun.created_at < (now - STALE_THRESHOLD), + ) + .all() + ) + for sr in stale_runs: + sr.status = "FAILED" + sr.error_message = "Stale: concurrency deadlock — marked by scheduled trigger" + sr.completed_at = now + db.commit() + logger.explore(f"Cleared {len(stale_runs)} stale PENDING run(s), proceeding with scheduled trigger", { + "job_id": job_id, + "stale_run_ids": [sr.id for sr in stale_runs], + }) + # Proceed to execute the scheduled run (do NOT return) + else: + logger.explore("Skipping scheduled run — concurrent run in progress", { + "job_id": job_id, + "existing_run_id": active_run.id, + }) + event_log = TranslationEventLog(db) + event_log.log_event( + job_id=job_id, + event_type="RUN_STARTED", + payload={"reason": "skipped_concurrent", "existing_run_id": active_run.id}, + ) + return # Check baseline expiry baseline_expired = False @@ -315,7 +354,17 @@ def execute_scheduled_translation( job_id=job_id, is_scheduled=True, ) - run.trigger_type = "scheduled" + + # Determine trigger type based on execution_mode and baseline expiry + if execution_mode == "new_key_only" and not baseline_expired: + run.trigger_type = "new_key_only" + elif execution_mode == "new_key_only" and baseline_expired: + logger.reason("Baseline expired — falling back to full translation", { + "job_id": job_id, "schedule_id": schedule_id, "age_days": age.days, + }) + run.trigger_type = "scheduled" + else: + run.trigger_type = "scheduled" db.flush() if baseline_expired: diff --git a/backend/src/schemas/settings.py b/backend/src/schemas/settings.py index 51de98bb..3d3f7dad 100644 --- a/backend/src/schemas/settings.py +++ b/backend/src/schemas/settings.py @@ -102,6 +102,7 @@ class TranslationScheduleItem(BaseModel): cron_expression: str timezone: str is_active: bool + execution_mode: str = "full" last_run_at: Optional[datetime] = None next_run_at: Optional[datetime] = None created_by: Optional[str] = None diff --git a/backend/src/schemas/translate.py b/backend/src/schemas/translate.py index 516f3462..ac9b12cf 100644 --- a/backend/src/schemas/translate.py +++ b/backend/src/schemas/translate.py @@ -313,6 +313,7 @@ class ScheduleConfig(BaseModel): cron_expression: str = Field(..., description="Cron expression for scheduling (e.g. '0 2 * * *')") timezone: str = Field("UTC", description="Timezone for the cron schedule (e.g. 'UTC', 'Europe/Moscow')") is_active: bool = True + execution_mode: str = Field("full", description="Translation execution mode: 'full' or 'new_key_only'") # #endregion ScheduleConfig @@ -324,6 +325,7 @@ class ScheduleResponse(BaseModel): cron_expression: str timezone: str is_active: bool + execution_mode: str = "full" last_run_at: Optional[datetime] = None next_run_at: Optional[datetime] = None created_by: Optional[str] = None diff --git a/backend/src/scripts/clean_release_cli.py b/backend/src/scripts/clean_release_cli.py index 9be69717..5eb846b5 100644 --- a/backend/src/scripts/clean_release_cli.py +++ b/backend/src/scripts/clean_release_cli.py @@ -10,6 +10,7 @@ import json from datetime import date, datetime, timezone from typing import Any, Dict, List, Optional +from ..core.cot_logger import seed_trace_id from ..models.clean_release import CandidateArtifact, ReleaseCandidate from ..services.clean_release.approval_service import ( approve_candidate, @@ -468,6 +469,7 @@ def run_revoke(args: argparse.Namespace) -> int: # #region main [TYPE Function] # @BRIEF CLI entrypoint for clean release commands. def main(argv: Optional[List[str]] = None) -> int: + seed_trace_id() parser = build_parser() args = parser.parse_args(argv) diff --git a/backend/src/scripts/clean_release_tui.py b/backend/src/scripts/clean_release_tui.py index 5b12d102..1cd820bc 100644 --- a/backend/src/scripts/clean_release_tui.py +++ b/backend/src/scripts/clean_release_tui.py @@ -19,6 +19,7 @@ BACKEND_ROOT = os.path.abspath(os.path.join(SCRIPT_DIR, "..", "..")) if BACKEND_ROOT not in sys.path: sys.path.insert(0, BACKEND_ROOT) +from src.core.cot_logger import seed_trace_id from src.models.clean_release import ( CandidateArtifact, CheckFinalStatus, @@ -662,6 +663,7 @@ def tui_main(stdscr: curses.window): def main() -> int: + seed_trace_id() # TUI requires interactive terminal; headless mode must use CLI/API flow. if not sys.stdout.isatty(): print( diff --git a/backend/src/scripts/create_admin.py b/backend/src/scripts/create_admin.py index 3b0a42a5..5420fb06 100644 --- a/backend/src/scripts/create_admin.py +++ b/backend/src/scripts/create_admin.py @@ -19,6 +19,7 @@ from src.core.database import AuthSessionLocal, init_db from src.core.auth.security import get_password_hash from src.models.auth import User, Role from src.core.logger import logger, belief_scope +from src.core.cot_logger import seed_trace_id # #region create_admin [TYPE Function] @@ -27,6 +28,7 @@ from src.core.logger import logger, belief_scope # @POST: Admin user exists in auth.db. # def create_admin(username, password, email=None): + seed_trace_id() with belief_scope("create_admin"): db = AuthSessionLocal() try: diff --git a/backend/src/scripts/init_auth_db.py b/backend/src/scripts/init_auth_db.py index 0117e429..fb45685d 100644 --- a/backend/src/scripts/init_auth_db.py +++ b/backend/src/scripts/init_auth_db.py @@ -17,6 +17,7 @@ sys.path.append(str(Path(__file__).parent.parent.parent)) from src.core.database import init_db from src.core.encryption_key import ensure_encryption_key from src.core.logger import logger, belief_scope +from src.core.cot_logger import seed_trace_id from src.scripts.seed_permissions import seed_permissions @@ -27,6 +28,7 @@ from src.scripts.seed_permissions import seed_permissions # @RELATION CALLS -> init_db # @RELATION CALLS -> seed_permissions def run_init(): + seed_trace_id() with belief_scope("init_auth_db"): logger.info("Initializing authentication database...") try: diff --git a/backend/src/scripts/migrate_sqlite_to_postgres.py b/backend/src/scripts/migrate_sqlite_to_postgres.py index 0bbd612f..27eac51f 100644 --- a/backend/src/scripts/migrate_sqlite_to_postgres.py +++ b/backend/src/scripts/migrate_sqlite_to_postgres.py @@ -21,6 +21,7 @@ from sqlalchemy import create_engine, text from sqlalchemy.exc import SQLAlchemyError from src.core.logger import belief_scope, logger +from src.core.cot_logger import seed_trace_id # #region Constants [TYPE Section] @@ -327,6 +328,7 @@ def _migrate_tasks_and_logs(engine, sqlite_conn: sqlite3.Connection) -> Dict[str def run_migration( sqlite_path: Path, target_url: str, legacy_config_path: Optional[Path] ) -> Dict[str, int]: + seed_trace_id() with belief_scope("run_migration"): logger.reason( f"sqlite={sqlite_path} target={target_url}", diff --git a/backend/src/scripts/seed_permissions.py b/backend/src/scripts/seed_permissions.py index 65e4590f..398bd35f 100644 --- a/backend/src/scripts/seed_permissions.py +++ b/backend/src/scripts/seed_permissions.py @@ -19,6 +19,7 @@ from src.core.database import AuthSessionLocal from src.models.auth import Permission, Role from src.core.auth.repository import AuthRepository from src.core.logger import logger, belief_scope +from src.core.cot_logger import seed_trace_id # #region INITIAL_PERMISSIONS [C:3] [TYPE Constant] # @BRIEF Canonical bootstrap permission tuples seeded into auth storage. @@ -66,6 +67,7 @@ INITIAL_PERMISSIONS = [ # @RELATION DEPENDS_ON -> AuthRepository # @RELATION DEPENDS_ON -> INITIAL_PERMISSIONS def seed_permissions(): + seed_trace_id() with belief_scope("seed_permissions"): db = AuthSessionLocal() try: diff --git a/backend/src/scripts/seed_superset_load_test.py b/backend/src/scripts/seed_superset_load_test.py index c067d987..c077ce57 100644 --- a/backend/src/scripts/seed_superset_load_test.py +++ b/backend/src/scripts/seed_superset_load_test.py @@ -19,6 +19,7 @@ sys.path.append(str(Path(__file__).parent.parent.parent)) from src.core.config_manager import ConfigManager from src.core.config_models import Environment from src.core.logger import belief_scope, logger +from src.core.cot_logger import seed_trace_id from src.core.superset_client import SupersetClient @@ -255,6 +256,7 @@ def _build_chart_template_pool( # @POST: Returns execution statistics dictionary. # @SIDE_EFFECT: Creates objects in Superset environments. def seed_superset_load_data(args: argparse.Namespace) -> Dict: + seed_trace_id() rng = random.Random(args.seed) env_map = _resolve_target_envs(args.envs) @@ -381,6 +383,7 @@ def seed_superset_load_data(args: argparse.Namespace) -> Dict: # @PRE: Command line arguments are valid. # @POST: Prints summary and exits with non-zero status on failure. def main() -> None: + seed_trace_id() with belief_scope("seed_superset_load_test.main"): args = _parse_args() result = seed_superset_load_data(args) diff --git a/backend/src/scripts/test_dataset_dashboard_relations.py b/backend/src/scripts/test_dataset_dashboard_relations.py index ccc9add1..5cdbb1c2 100644 --- a/backend/src/scripts/test_dataset_dashboard_relations.py +++ b/backend/src/scripts/test_dataset_dashboard_relations.py @@ -18,10 +18,12 @@ sys.path.append(str(Path(__file__).parent.parent.parent)) from src.core.superset_client import SupersetClient from src.core.config_manager import ConfigManager from src.core.logger import logger +from src.core.cot_logger import seed_trace_id def test_dashboard_dataset_relations(): """Test fetching dataset-to-dashboard relationships.""" + seed_trace_id() # Load environment from existing config config_manager = ConfigManager() diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 00000000..27d51e88 --- /dev/null +++ b/backend/tests/conftest.py @@ -0,0 +1,25 @@ +# #region TestSessionConfig [C:2] [TYPE Module] [SEMANTICS test, conftest, db] +# @BRIEF Shared pytest fixtures and session configuration for backend tests. +# @RELATION BINDS_TO -> [init_db] +# @PRE All test modules use sys.path patching to import from src/. +# @POST Database tables exist before test session executes. +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest + + +@pytest.fixture(scope="session", autouse=True) +def ensure_db_tables(): + """Ensure all tables exist before test session. + + This fixture runs once per test session, creating all registered + SQLAlchemy tables. Individual test modules may also create their + own in-memory engines — this guarantees the real engine schema + is initialized for tests that depend on it. + """ + from src.core.database import init_db + init_db() +# #endregion TestSessionConfig diff --git a/backend/tests/test_core_scheduler.py b/backend/tests/test_core_scheduler.py new file mode 100644 index 00000000..dcb59476 --- /dev/null +++ b/backend/tests/test_core_scheduler.py @@ -0,0 +1,177 @@ +# #region TestCoreScheduler [C:3] [TYPE Module] [SEMANTICS test, scheduler, apscheduler, load, reload] +# @BRIEF Verify SchedulerService load_schedules and translation job add/remove contracts. +# @RELATION BINDS_TO -> [SchedulerService] +# @TEST_EDGE: load_schedules_reloads — jobs survive remove_all_jobs when re-registered from DB +# @TEST_EDGE: add_remove_translation_job — job correctly registered and removed from APScheduler +# @TEST_INVARIANT: translation_jobs_persist_across_reload -> VERIFIED_BY: [test_load_schedules_reloads_translation_schedules] + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest +from unittest.mock import MagicMock, patch +from datetime import datetime, timezone +from apscheduler.schedulers.background import BackgroundScheduler + + +# --------------------------------------------------------------------------- +# Helpers — fresh scheduler & mock config +# --------------------------------------------------------------------------- + +def _make_mock_config_manager(backup_env_id="env1", backup_cron="0 3 * * *"): + """Build a mock ConfigManager returning a single backup-enabled environment.""" + mgr = MagicMock() + mock_env = MagicMock() + mock_env.id = backup_env_id + mock_env.backup_schedule = MagicMock() + mock_env.backup_schedule.enabled = True + mock_env.backup_schedule.cron_expression = backup_cron + mock_config = MagicMock() + mock_config.environments = [mock_env] + mgr.get_config.return_value = mock_config + return mgr + + +def _make_mock_task_manager(): + """Build a mock TaskManager returning no active tasks.""" + tm = MagicMock() + tm.get_tasks.return_value = [] + return tm + + +# --------------------------------------------------------------------------- +# Fixture helpers — start/stop a real BackgroundScheduler +# --------------------------------------------------------------------------- + +def _start_scheduler(): + """Create and start a BackgroundScheduler. Returns the scheduler instance.""" + sched = BackgroundScheduler() + sched.start() + return sched + + +def _stop_scheduler(sched): + """Shutdown a BackgroundScheduler if running.""" + if sched.running: + sched.shutdown(wait=False) + + +# #region test_load_schedules_reloads_translation_schedules [C:1] [TYPE Function] +def test_load_schedules_reloads_translation_schedules(): + """Verify load_schedules re-registers translation jobs after remove_all_jobs.""" + from src.core.scheduler import SchedulerService + + sched = _start_scheduler() + try: + cfg = _make_mock_config_manager() + tm = _make_mock_task_manager() + + svc = SchedulerService.__new__(SchedulerService) + svc.task_manager = tm + svc.config_manager = cfg + svc.scheduler = sched + + # Pre-register a translation job (simulates what happens before reload) + svc.add_translation_job( + schedule_id="sched-1", + job_id="job-1", + cron_expression="0 2 * * *", + ) + + # Verify it exists BEFORE reload + job_before = sched.get_job("translate_sched-1") + assert job_before is not None, "Translation job should exist before reload" + + # Mock the DB query that load_schedules uses internally + mock_schedule = MagicMock() + mock_schedule.id = "sched-1" + mock_schedule.job_id = "job-1" + mock_schedule.cron_expression = "0 2 * * *" + mock_schedule.timezone = "UTC" + mock_schedule.execution_mode = "full" + + mock_db = MagicMock() + mock_db.query.return_value.filter.return_value.all.return_value = [mock_schedule] + + with patch("src.core.scheduler.SessionLocal", return_value=mock_db): + svc.load_schedules() + + # After reload, the translation job should STILL exist + job_after = sched.get_job("translate_sched-1") + assert job_after is not None, ( + "Translation job should survive load_schedules (re-registered from DB)" + ) + + # Backup job should also exist (added from config) + backup_job = sched.get_job("backup_env1") + assert backup_job is not None, "Backup job should be registered by load_schedules" + finally: + _stop_scheduler(sched) +# #endregion test_load_schedules_reloads_translation_schedules + + +# #region test_add_and_remove_translation_job [C:1] [TYPE Function] +def test_add_and_remove_translation_job(): + """Verify add_translation_job registers and remove_translation_job removes.""" + from src.core.scheduler import SchedulerService + + sched = _start_scheduler() + try: + cfg = _make_mock_config_manager() + tm = _make_mock_task_manager() + + svc = SchedulerService.__new__(SchedulerService) + svc.task_manager = tm + svc.config_manager = cfg + svc.scheduler = sched + + # Should NOT exist before add + assert sched.get_job("translate_sched-x") is None + + svc.add_translation_job( + schedule_id="sched-x", + job_id="job-x", + cron_expression="*/5 * * * *", + ) + + # Should exist after add + job = sched.get_job("translate_sched-x") + assert job is not None, "Translation job should be registered after add" + assert job.id == "translate_sched-x" + + svc.remove_translation_job("sched-x") + + # Should NOT exist after remove + assert sched.get_job("translate_sched-x") is None, ( + "Translation job should be removed after remove_translation_job" + ) + finally: + _stop_scheduler(sched) +# #endregion test_add_and_remove_translation_job + + +# #region test_remove_nonexistent_translation_job_no_error [C:1] [TYPE Function] +def test_remove_nonexistent_translation_job_no_error(): + """Verify remove_translation_job handles non-existent job silently.""" + from src.core.scheduler import SchedulerService + + sched = _start_scheduler() + try: + cfg = _make_mock_config_manager() + tm = _make_mock_task_manager() + + svc = SchedulerService.__new__(SchedulerService) + svc.task_manager = tm + svc.config_manager = cfg + svc.scheduler = sched + + # Should not raise — silent ignore + svc.remove_translation_job("nonexistent-sched") + finally: + _stop_scheduler(sched) +# #endregion test_remove_nonexistent_translation_job_no_error + + +# #endregion TestCoreScheduler diff --git a/backend/tests/test_translate_executor_filter.py b/backend/tests/test_translate_executor_filter.py new file mode 100644 index 00000000..11d5d1d8 --- /dev/null +++ b/backend/tests/test_translate_executor_filter.py @@ -0,0 +1,291 @@ +# #region TestTranslateExecutorFilter [C:3] [TYPE Module] [SEMANTICS test, translate, executor, filter, new-key-only] +# @BRIEF Verify TranslationExecutor._filter_new_keys contracts — key dedup, empty-set, guard triggers. +# @RELATION BINDS_TO -> [TranslationExecutor] +# @TEST_EDGE: no_prior_run — all rows pass through when no previous run exists +# @TEST_EDGE: filter_existing_keys — rows with keys from prior run are skipped +# @TEST_EDGE: composite_keys — multiple target_key_cols produce correct dedup +# @TEST_EDGE: all_existing — zero new rows returns empty list +# @TEST_EDGE: empty_key_cols — guard triggers, all rows returned with warning +# @TEST_EDGE: null_source_data — None source_data is safely handled +# @TEST_INVARIANT: new_key_only_filter -> VERIFIED_BY: [test_no_prior_run_returns_all_rows, test_filters_existing_keys, test_composite_keys] + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest +from unittest.mock import MagicMock + +from src.plugins.translate.executor import TranslationExecutor + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_executor(mock_db=None): + """Create a TranslationExecutor with a mocked DB session and config.""" + db = mock_db or MagicMock() + cfg = MagicMock() + return TranslationExecutor(db, cfg, current_user="test_user") + + +def _make_job(source_key_cols=None, target_key_cols=None): + """Build a MagicMock TranslationJob with key columns configured.""" + job = MagicMock() + job.id = "job-test-1" + job.source_key_cols = source_key_cols or [] + job.target_key_cols = target_key_cols or [] + return job + + +def _make_rows(*source_data_dicts): + """Build source_rows list from source_data dicts.""" + rows = [] + for sd in source_data_dicts: + rows.append({ + "row_index": sd.get("_idx", "0"), + "source_text": sd.get("text", ""), + "source_data": sd, + }) + return rows + + +def _make_prev_records(run_id, source_data_dicts): + """Build MagicMock TranslationRecord list with source_data and SUCCESS status.""" + recs = [] + for sd in source_data_dicts: + rec = MagicMock() + rec.run_id = run_id + rec.status = "SUCCESS" + rec.source_data = sd + recs.append(rec) + return recs + + +def _wire_query_chain(mock_db, prev_run, prev_records): + """Wire mock DB for _filter_new_keys: two sequential db.query() calls. + + Query 1: db.query(TranslationRun).filter(a,b,c,d).order_by(desc).first() → prev_run + Query 2: db.query(TranslationRecord).filter(a,b).all() → prev_records + """ + q1 = MagicMock() + q1.filter.return_value.order_by.return_value.first.return_value = prev_run + q2 = MagicMock() + q2.filter.return_value.all.return_value = prev_records + mock_db.query.side_effect = [q1, q2] + + +# #region test_no_prior_run_returns_all_rows [C:1] [TYPE Function] +def test_no_prior_run_returns_all_rows(): + """When no prior TranslationRun exists, all source_rows pass through unchanged.""" + executor = _make_executor() + job = _make_job(source_key_cols=["id"], target_key_cols=["id"]) + + rows = _make_rows( + {"id": 1, "text": "hello", "_idx": "0"}, + {"id": 2, "text": "world", "_idx": "1"}, + {"id": 3, "text": "test", "_idx": "2"}, + ) + + # Mock: no prior run (query returns None) + _wire_query_chain(executor.db, prev_run=None, prev_records=[]) + + result = executor._filter_new_keys(job, "run-1", rows) + assert len(result) == 3 + assert result == rows # identity — all pass through +# #endregion test_no_prior_run_returns_all_rows + + +# #region test_filters_existing_keys [C:1] [TYPE Function] +def test_filters_existing_keys(): + """Rows whose key exists in a prior run's records are excluded.""" + executor = _make_executor() + job = _make_job(source_key_cols=["id"], target_key_cols=["id"]) + + rows = _make_rows( + {"id": 1, "text": "existing", "_idx": "0"}, + {"id": 2, "text": "new", "_idx": "1"}, + {"id": 3, "text": "existing", "_idx": "2"}, + ) + + mock_prev_run = MagicMock() + mock_prev_run.id = "prev-run-1" + + # Only id=1 and id=3 were previously translated successfully + prev_recs = _make_prev_records("prev-run-1", [ + {"id": 1, "text": "existing"}, + {"id": 3, "text": "existing"}, + ]) + + _wire_query_chain(executor.db, mock_prev_run, prev_recs) + + result = executor._filter_new_keys(job, "run-1", rows) + assert len(result) == 1, f"Expected 1 new row, got {len(result)}" + assert result[0]["source_data"]["id"] == 2 +# #endregion test_filters_existing_keys + + +# #region test_composite_keys [C:1] [TYPE Function] +def test_composite_keys(): + """Multiple target_key_cols produce correct composite key dedup.""" + executor = _make_executor() + job = _make_job(target_key_cols=["schema", "table"]) + + rows = _make_rows( + {"schema": "public", "table": "users", "text": "a", "_idx": "0"}, + {"schema": "public", "table": "orders", "text": "b", "_idx": "1"}, + {"schema": "admin", "table": "users", "text": "c", "_idx": "2"}, + ) + + mock_prev_run = MagicMock() + mock_prev_run.id = "prev-run-1" + + # Only (public, users) was already translated + prev_recs = _make_prev_records("prev-run-1", [ + {"schema": "public", "table": "users"}, + ]) + + _wire_query_chain(executor.db, mock_prev_run, prev_recs) + + result = executor._filter_new_keys(job, "run-1", rows) + assert len(result) == 2, f"Expected 2 new rows, got {len(result)}" + + result_keys = { + tuple(str(r["source_data"].get(k, "")) for k in ["schema", "table"]) + for r in result + } + assert ("public", "orders") in result_keys + assert ("admin", "users") in result_keys + assert ("public", "users") not in result_keys +# #endregion test_composite_keys + + +# #region test_all_keys_existing_returns_empty [C:1] [TYPE Function] +def test_all_keys_existing_returns_empty(): + """When every source row's key already exists, the filtered list is empty.""" + executor = _make_executor() + job = _make_job(source_key_cols=["id"]) + + rows = _make_rows( + {"id": 1, "text": "a", "_idx": "0"}, + {"id": 2, "text": "b", "_idx": "1"}, + ) + + mock_prev_run = MagicMock() + mock_prev_run.id = "prev-run-1" + + prev_recs = _make_prev_records("prev-run-1", [ + {"id": 1}, {"id": 2}, + ]) + + _wire_query_chain(executor.db, mock_prev_run, prev_recs) + + result = executor._filter_new_keys(job, "run-1", rows) + assert result == [], "All keys existing should produce empty result" +# #endregion test_all_keys_existing_returns_empty + + +# #region test_empty_key_cols_returns_all_rows [C:1] [TYPE Function] +def test_empty_key_cols_returns_all_rows(): + """When no key columns are configured, guard triggers — all rows returned.""" + executor = _make_executor() + job = _make_job(source_key_cols=[], target_key_cols=[]) + + rows = _make_rows( + {"id": 1, "text": "a", "_idx": "0"}, + {"id": 2, "text": "b", "_idx": "1"}, + ) + + mock_prev_run = MagicMock() + mock_prev_run.id = "prev-run-1" + + _wire_query_chain(executor.db, mock_prev_run, prev_records=[]) + + # Even with a prior run and records, no key cols means all rows pass through + result = executor._filter_new_keys(job, "run-1", rows) + assert len(result) == 2 + assert result == rows +# #endregion test_empty_key_cols_returns_all_rows + + +# #region test_null_source_data_handled [C:1] [TYPE Function] +def test_null_source_data_handled(): + """None source_data is safely handled with or {} fallback — no crash.""" + executor = _make_executor() + job = _make_job(source_key_cols=["id"]) + + rows = [ + {"row_index": "0", "source_text": "a", "source_data": None}, + {"row_index": "1", "source_text": "b", "source_data": {"id": 2}}, + ] + + mock_prev_run = MagicMock() + mock_prev_run.id = "prev-run-1" + + # Previous run only had id=2 + prev_recs = _make_prev_records("prev-run-1", [{"id": 2}]) + + _wire_query_chain(executor.db, mock_prev_run, prev_recs) + + # Should NOT crash on None source_data + result = executor._filter_new_keys(job, "run-1", rows) + + # The row with None source_data has key_tuple ("",) — NOT in existing_keys + # (existing has ("2",)). So it should pass through. + assert len(result) == 1, ( + f"Expected 1 new row (None source_data treated as ''), got {len(result)}" + ) + assert result[0]["source_data"] is None +# #endregion test_null_source_data_handled + + +# #region test_prior_run_no_records_returns_all [C:1] [TYPE Function] +def test_prior_run_no_records_returns_all(): + """Prior run exists but has no successful records — all rows treated as new.""" + executor = _make_executor() + job = _make_job(source_key_cols=["id"]) + + rows = _make_rows( + {"id": 1, "text": "a", "_idx": "0"}, + {"id": 2, "text": "b", "_idx": "1"}, + ) + + mock_prev_run = MagicMock() + mock_prev_run.id = "prev-run-1" + + _wire_query_chain(executor.db, mock_prev_run, prev_records=[]) + + result = executor._filter_new_keys(job, "run-1", rows) + assert len(result) == 2 + assert result == rows +# #endregion test_prior_run_no_records_returns_all + + +# #region test_key_cols_none_returns_all [C:1] [TYPE Function] +def test_key_cols_none_returns_all(): + """When source_key_cols and target_key_cols are both None, all rows pass through.""" + executor = _make_executor() + # Job with no key_cols at all (None, not empty list) + job = MagicMock() + job.id = "job-test-1" + job.source_key_cols = None + job.target_key_cols = None + + rows = _make_rows( + {"id": 1, "text": "a", "_idx": "0"}, + ) + + mock_prev_run = MagicMock() + mock_prev_run.id = "prev-run-1" + + _wire_query_chain(executor.db, mock_prev_run, prev_records=[]) + + result = executor._filter_new_keys(job, "run-1", rows) + assert len(result) == 1 + assert result == rows +# #endregion test_key_cols_none_returns_all + + +# #endregion TestTranslateExecutorFilter diff --git a/backend/tests/test_translate_history.py b/backend/tests/test_translate_history.py index 577a71b0..697430c2 100644 --- a/backend/tests/test_translate_history.py +++ b/backend/tests/test_translate_history.py @@ -197,7 +197,7 @@ def test_list_runs_filter_status(client, mock_api_deps): _create_test_run(session, job.id, status="FAILED") session.commit() - response = client.get(f"/api/translate/runs?status=FAILED") + response = client.get(f"/api/translate/runs?run_status=FAILED") assert response.status_code == 200 data = response.json() for item in data["items"]: diff --git a/backend/tests/test_translate_scheduler_execution.py b/backend/tests/test_translate_scheduler_execution.py new file mode 100644 index 00000000..5f64cc70 --- /dev/null +++ b/backend/tests/test_translate_scheduler_execution.py @@ -0,0 +1,399 @@ +# #region TestTranslateSchedulerExecution [C:3] [TYPE Module] [SEMANTICS test, translate, scheduler, execution, trigger-type, baseline] +# @BRIEF Verify execute_scheduled_translation contracts — trigger_type dispatch, baseline expiry fallback. +# @RELATION BINDS_TO -> [execute_scheduled_translation] +# @TEST_EDGE: new_key_only_mode — trigger_type="new_key_only" when baseline not expired +# @TEST_EDGE: baseline_expired_fallback — trigger_type="scheduled" when baseline > 90 days old +# @TEST_EDGE: full_mode_default — trigger_type="scheduled" for execution_mode="full" +# @TEST_EDGE: inactive_schedule — function returns early without executing +# @TEST_EDGE: execution_error_handled — run marked FAILED, schedule tracking updated +# @TEST_INVARIANT: trigger_type_dispatch -> VERIFIED_BY: [test_new_key_only_mode, test_baseline_expired_fallback, test_full_mode_default] +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) +from unittest.mock import MagicMock, patch +from datetime import datetime, timezone, timedelta + + +# -- Helpers ----------------------------------------------------------------- + +def _make_most_recent_run(created_at_delta_days=60): + """Build a MagicMock TranslationRun with given age.""" + run = MagicMock() + run.id = "prev-run-1" + run.job_id = "job-1" + run.status = "COMPLETED" + run.insert_status = "succeeded" + run.created_at = datetime.now(timezone.utc) - timedelta(days=created_at_delta_days) + return run + + +def _make_mock_run(run_id="run-1", job_id="job-1"): + """Build a mock TranslationRun returned by start_run.""" + run = MagicMock() + run.id = run_id + run.job_id = job_id + run.status = "PENDING" + run.trigger_type = None + run.total_records = 0 + run.successful_records = 0 + run.failed_records = 0 + run.skipped_records = 0 + run.insert_status = None + run.superset_execution_id = None + run.superset_execution_log = None + run.error_message = None + return run + + +# -- Tests: trigger_type dispatch and execution modes ----------------------- + +# #region test_new_key_only_mode [C:1] [TYPE Function] +def test_new_key_only_mode(): + """execution_mode="new_key_only" with 60-day-old prev run → trigger_type="new_key_only".""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = True + schedule.execution_mode = "new_key_only" + + most_recent = _make_most_recent_run(created_at_delta_days=60) # 60 days < 90 + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + q2 = MagicMock() + q2.filter.return_value.order_by.return_value.first.return_value = None # no concurrent + + q3 = MagicMock() + q3.filter.return_value.order_by.return_value.first.return_value = most_recent + + mock_db.query.side_effect = [q1, q2, q3] + + mock_run = _make_mock_run() + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + mock_orch = MagicMock() + mock_orch.start_run.return_value = mock_run + mock_orch_cls.return_value = mock_orch + + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="new_key_only", + ) + + assert mock_run.trigger_type == "new_key_only", ( + f"Expected trigger_type='new_key_only', got '{mock_run.trigger_type}'" + ) + mock_orch.start_run.assert_called_once_with(job_id="job-1", is_scheduled=True) + mock_orch.execute_run.assert_called_once() + mock_db.close.assert_called_once() +# #endregion test_new_key_only_mode + + +# #region test_baseline_expired_fallback [C:1] [TYPE Function] +def test_baseline_expired_fallback(): + """execution_mode="new_key_only" with >90-day-old prev run → trigger_type="scheduled" baseline fallback.""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = True + schedule.execution_mode = "new_key_only" + + most_recent = _make_most_recent_run(created_at_delta_days=95) # 95 days > 90 + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + q2 = MagicMock() + q2.filter.return_value.order_by.return_value.first.return_value = None + + q3 = MagicMock() + q3.filter.return_value.order_by.return_value.first.return_value = most_recent + + mock_db.query.side_effect = [q1, q2, q3] + + mock_run = _make_mock_run() + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + mock_orch = MagicMock() + mock_orch.start_run.return_value = mock_run + mock_orch_cls.return_value = mock_orch + + with patch( + "src.plugins.translate.scheduler.TranslationEventLog" + ) as mock_event_log_cls: + mock_event_log = MagicMock() + mock_event_log_cls.return_value = mock_event_log + + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="new_key_only", + ) + + assert mock_run.trigger_type == "scheduled", ( + f"Expected trigger_type='scheduled' (baseline expired), got '{mock_run.trigger_type}'" + ) + + mock_event_log.log_event.assert_called_once() + event_call = mock_event_log.log_event.call_args + assert event_call.kwargs["job_id"] == "job-1" + assert event_call.kwargs["event_type"] == "RUN_STARTED" + assert event_call.kwargs["payload"]["reason"] == "baseline_expired" + assert event_call.kwargs["payload"]["age_days"] == 95 + + mock_db.close.assert_called_once() +# #endregion test_baseline_expired_fallback + + +# #region test_full_mode_default [C:1] [TYPE Function] +def test_full_mode_default(): + """execution_mode="full" → trigger_type="scheduled" regardless of baseline age.""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = True + schedule.execution_mode = "full" + + most_recent = _make_most_recent_run(created_at_delta_days=30) + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + q2 = MagicMock() + q2.filter.return_value.order_by.return_value.first.return_value = None + + q3 = MagicMock() + q3.filter.return_value.order_by.return_value.first.return_value = most_recent + + mock_db.query.side_effect = [q1, q2, q3] + + mock_run = _make_mock_run() + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + mock_orch = MagicMock() + mock_orch.start_run.return_value = mock_run + mock_orch_cls.return_value = mock_orch + + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="full", + ) + + assert mock_run.trigger_type == "scheduled", ( + f"Expected trigger_type='scheduled' for full mode, got '{mock_run.trigger_type}'" + ) + mock_orch.execute_run.assert_called_once() + mock_db.close.assert_called_once() +# #endregion test_full_mode_default + + +# #region test_inactive_schedule_skips [C:1] [TYPE Function] +def test_inactive_schedule_skips(): + """Inactive schedule → function returns early without executing anything.""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = False + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + mock_db.query.return_value = q1 + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="full", + ) + + mock_orch_cls.assert_not_called() + mock_db.close.assert_called_once() +# #endregion test_inactive_schedule_skips + + +# #region test_schedule_not_found_skips [C:1] [TYPE Function] +def test_schedule_not_found_skips(): + """Schedule not found → function returns early.""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + q1 = MagicMock() + q1.filter.return_value.first.return_value = None + + mock_db.query.return_value = q1 + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="full", + ) + + mock_orch_cls.assert_not_called() + mock_db.close.assert_called_once() +# #endregion test_schedule_not_found_skips + + +# #region test_baseline_expired_full_mode [C:1] [TYPE Function] +def test_baseline_expired_full_mode(): + """execution_mode="full" with expired baseline → trigger_type stays 'scheduled'.""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = True + schedule.execution_mode = "full" + + most_recent = _make_most_recent_run(created_at_delta_days=100) # expired + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + q2 = MagicMock() + q2.filter.return_value.order_by.return_value.first.return_value = None + + q3 = MagicMock() + q3.filter.return_value.order_by.return_value.first.return_value = most_recent + + mock_db.query.side_effect = [q1, q2, q3] + + mock_run = _make_mock_run() + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + mock_orch = MagicMock() + mock_orch.start_run.return_value = mock_run + mock_orch_cls.return_value = mock_orch + + with patch( + "src.plugins.translate.scheduler.TranslationEventLog" + ) as mock_event_log_cls: + mock_event_log = MagicMock() + mock_event_log_cls.return_value = mock_event_log + + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="full", + ) + + assert mock_run.trigger_type == "scheduled" + mock_event_log.log_event.assert_called_once() + event_call = mock_event_log.log_event.call_args + assert event_call.kwargs["payload"]["reason"] == "baseline_expired" + mock_db.close.assert_called_once() +# #endregion test_baseline_expired_full_mode + + +# #region test_execution_error_handled [C:1] [TYPE Function] +def test_execution_error_handled(): + """execute_run raises → run is marked FAILED, schedule tracking updated, no crash.""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = True + schedule.execution_mode = "full" + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + q2 = MagicMock() + q2.filter.return_value.order_by.return_value.first.return_value = None + + q3 = MagicMock() + q3.filter.return_value.order_by.return_value.first.return_value = None + + mock_db.query.side_effect = [q1, q2, q3] + + mock_run = _make_mock_run() + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + mock_orch = MagicMock() + mock_orch.start_run.return_value = mock_run + mock_orch.execute_run.side_effect = RuntimeError("LLM timeout") + mock_orch_cls.return_value = mock_orch + + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="full", + ) + + assert mock_run.status == "FAILED" + assert "LLM timeout" in mock_run.error_message + assert mock_run.completed_at is not None + assert schedule.last_run_at is not None + mock_db.commit.assert_called() + mock_db.close.assert_called_once() +# #endregion test_execution_error_handled + +# #endregion TestTranslateSchedulerExecution diff --git a/backend/tests/test_translate_scheduler_guard.py b/backend/tests/test_translate_scheduler_guard.py new file mode 100644 index 00000000..ee0b8ed2 --- /dev/null +++ b/backend/tests/test_translate_scheduler_guard.py @@ -0,0 +1,302 @@ +# #region TestTranslateSchedulerGuard [C:3] [TYPE Module] [SEMANTICS test, translate, scheduler, guard, concurrency, stale, pending] +# @BRIEF Verify execute_scheduled_translation concurrency guard and stale-PENDING clearance contracts. +# @RELATION BINDS_TO -> [execute_scheduled_translation] +# @TEST_EDGE: concurrent_run_skips — active RUNNING concurrent run → function skips and logs skipped_concurrent event +# @TEST_EDGE: recent_pending_run_not_stale — recent PENDING run (30 min old) → function skips, not stale +# @TEST_EDGE: stale_pending_cleared_and_proceeds — stale PENDING run (>1h old) → cleared as FAILED, execution proceeds +# @TEST_EDGE: multiple_stale_pending_all_cleared — multiple stale PENDING runs → all cleared as FAILED, execution proceeds +# @TEST_INVARIANT: concurrency_guard -> VERIFIED_BY: [test_concurrent_run_skips, test_recent_pending_run_not_stale] +# @TEST_INVARIANT: stale_pending_clearance -> VERIFIED_BY: [test_stale_pending_cleared_and_proceeds, test_multiple_stale_pending_all_cleared] + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest +from unittest.mock import MagicMock, patch +from datetime import datetime, timezone, timedelta + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_most_recent_run(created_at_delta_days=60): + """Build a MagicMock TranslationRun with given age.""" + run = MagicMock() + run.id = "prev-run-1" + run.job_id = "job-1" + run.status = "COMPLETED" + run.insert_status = "succeeded" + run.created_at = datetime.now(timezone.utc) - timedelta(days=created_at_delta_days) + return run + + +def _make_mock_run(run_id="run-1", job_id="job-1"): + """Build a mock TranslationRun returned by start_run.""" + run = MagicMock() + run.id = run_id + run.job_id = job_id + run.status = "PENDING" + run.trigger_type = None + run.total_records = 0 + run.successful_records = 0 + run.failed_records = 0 + run.skipped_records = 0 + run.insert_status = None + run.superset_execution_id = None + run.superset_execution_log = None + run.error_message = None + return run + + +# --------------------------------------------------------------------------- +# Tests — concurrency guard and stale PENDING clearance +# --------------------------------------------------------------------------- + +# #region test_concurrent_run_skips [C:1] [TYPE Function] +def test_concurrent_run_skips(): + """Active RUNNING concurrent run → function skips and logs skipped_concurrent event.""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = True + schedule.execution_mode = "full" + + active_run = MagicMock() + active_run.id = "active-run-1" + active_run.job_id = "job-1" + active_run.status = "RUNNING" + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + q2 = MagicMock() + q2.filter.return_value.order_by.return_value.first.return_value = active_run + + mock_db.query.side_effect = [q1, q2] + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + with patch( + "src.plugins.translate.scheduler.TranslationEventLog" + ) as mock_event_log_cls: + mock_event_log = MagicMock() + mock_event_log_cls.return_value = mock_event_log + + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="full", + ) + + mock_orch_cls.assert_not_called() + mock_event_log.log_event.assert_called_once() + event_call = mock_event_log.log_event.call_args + assert event_call.kwargs["event_type"] == "RUN_STARTED" + assert event_call.kwargs["payload"]["reason"] == "skipped_concurrent" + mock_db.close.assert_called_once() +# #endregion test_concurrent_run_skips + + +# #region test_recent_pending_run_not_stale [C:1] [TYPE Function] +def test_recent_pending_run_not_stale(): + """Recent PENDING run (30 min old) → function skips, not stale.""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = True + schedule.execution_mode = "full" + + pending_run = MagicMock() + pending_run.id = "pending-run-1" + pending_run.job_id = "job-1" + pending_run.status = "PENDING" + pending_run.created_at = datetime.now(timezone.utc) - timedelta(minutes=30) + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + q2 = MagicMock() + q2.filter.return_value.order_by.return_value.first.return_value = pending_run + + mock_db.query.side_effect = [q1, q2] + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + with patch( + "src.plugins.translate.scheduler.TranslationEventLog" + ) as mock_event_log_cls: + mock_event_log = MagicMock() + mock_event_log_cls.return_value = mock_event_log + + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="full", + ) + + mock_orch_cls.assert_not_called() + mock_event_log.log_event.assert_called_once() + event_call = mock_event_log.log_event.call_args + assert event_call.kwargs["event_type"] == "RUN_STARTED" + assert event_call.kwargs["payload"]["reason"] == "skipped_concurrent" + mock_db.close.assert_called_once() +# #endregion test_recent_pending_run_not_stale + + +# #region test_stale_pending_cleared_and_proceeds [C:1] [TYPE Function] +def test_stale_pending_cleared_and_proceeds(): + """Stale PENDING run (>1h old) → cleared as FAILED, execution proceeds.""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = True + schedule.execution_mode = "full" + + stale_run = MagicMock() + stale_run.id = "stale-run-1" + stale_run.job_id = "job-1" + stale_run.status = "PENDING" + stale_run.created_at = datetime.now(timezone.utc) - timedelta(hours=2) + + most_recent = _make_most_recent_run(created_at_delta_days=30) + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + q2 = MagicMock() + q2.filter.return_value.order_by.return_value.first.return_value = stale_run + + q3 = MagicMock() + q3.filter.return_value.all.return_value = [stale_run] + + q4 = MagicMock() + q4.filter.return_value.order_by.return_value.first.return_value = most_recent + + mock_db.query.side_effect = [q1, q2, q3, q4] + + mock_run = _make_mock_run() + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + mock_orch = MagicMock() + mock_orch.start_run.return_value = mock_run + mock_orch_cls.return_value = mock_orch + + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="full", + ) + + assert stale_run.status == "FAILED" + assert "Stale: concurrency deadlock" in stale_run.error_message + assert stale_run.completed_at is not None + + mock_orch.start_run.assert_called_once_with(job_id="job-1", is_scheduled=True) + mock_orch.execute_run.assert_called_once() + assert mock_db.commit.call_count >= 1 + mock_db.close.assert_called_once() +# #endregion test_stale_pending_cleared_and_proceeds + + +# #region test_multiple_stale_pending_all_cleared [C:1] [TYPE Function] +def test_multiple_stale_pending_all_cleared(): + """Multiple stale PENDING runs → all cleared as FAILED, execution proceeds.""" + from src.plugins.translate.scheduler import execute_scheduled_translation + + mock_db = MagicMock() + mock_session_maker = MagicMock(return_value=mock_db) + mock_config = MagicMock() + + schedule = MagicMock() + schedule.id = "sched-1" + schedule.job_id = "job-1" + schedule.is_active = True + schedule.execution_mode = "full" + + stale_run_a = MagicMock() + stale_run_a.id = "stale-run-a" + stale_run_a.job_id = "job-1" + stale_run_a.status = "PENDING" + stale_run_a.created_at = datetime.now(timezone.utc) - timedelta(hours=5) + + stale_run_b = MagicMock() + stale_run_b.id = "stale-run-b" + stale_run_b.job_id = "job-1" + stale_run_b.status = "PENDING" + stale_run_b.created_at = datetime.now(timezone.utc) - timedelta(hours=3) + + most_recent = _make_most_recent_run(created_at_delta_days=30) + + q1 = MagicMock() + q1.filter.return_value.first.return_value = schedule + + q2 = MagicMock() + q2.filter.return_value.order_by.return_value.first.return_value = stale_run_a + + q3 = MagicMock() + q3.filter.return_value.all.return_value = [stale_run_a, stale_run_b] + + q4 = MagicMock() + q4.filter.return_value.order_by.return_value.first.return_value = most_recent + + mock_db.query.side_effect = [q1, q2, q3, q4] + + mock_run = _make_mock_run() + + with patch( + "src.plugins.translate.orchestrator.TranslationOrchestrator" + ) as mock_orch_cls: + mock_orch = MagicMock() + mock_orch.start_run.return_value = mock_run + mock_orch_cls.return_value = mock_orch + + execute_scheduled_translation( + schedule_id="sched-1", + job_id="job-1", + db_session_maker=mock_session_maker, + config_manager=mock_config, + execution_mode="full", + ) + + for sr in [stale_run_a, stale_run_b]: + assert sr.status == "FAILED" + assert "Stale: concurrency deadlock" in sr.error_message + assert sr.completed_at is not None + + mock_orch.start_run.assert_called_once_with(job_id="job-1", is_scheduled=True) + mock_orch.execute_run.assert_called_once() + mock_db.close.assert_called_once() +# #endregion test_multiple_stale_pending_all_cleared + +# #endregion TestTranslateSchedulerGuard diff --git a/frontend/src/lib/components/translate/ScheduleConfig.svelte b/frontend/src/lib/components/translate/ScheduleConfig.svelte index 9db7750c..882277bf 100644 --- a/frontend/src/lib/components/translate/ScheduleConfig.svelte +++ b/frontend/src/lib/components/translate/ScheduleConfig.svelte @@ -33,6 +33,7 @@ let uxState = $state('idle'); let cronExpression = $state('0 2 * * *'); let timezone = $state('UTC'); + let executionMode = $state('full'); let isEnabled = $state(false); let isLoading = $state(true); let nextExecutions = $state([]); @@ -50,6 +51,7 @@ if (sch && sch.cron_expression) { cronExpression = sch.cron_expression; timezone = sch.timezone || 'UTC'; + executionMode = sch.execution_mode || 'full'; isEnabled = sch.is_active !== false; scheduleExists = true; uxState = isEnabled ? 'enabled' : 'disabled'; @@ -80,6 +82,7 @@ cron_expression: cronExpression, timezone: timezone, is_active: isEnabled, + execution_mode: executionMode, }; await setSchedule(jobId, payload); addToast($t.translate?.schedule?.saved || 'Schedule saved', 'success'); @@ -218,6 +221,27 @@ + +
+ +
+ + +
+
+ {#if nextExecutions.length > 0}