refactor: govern background job execution

backend/bootstrap/container.py

@@ -9,6 +9,7 @@ from backend.database.repositories import RepositoryBundle, build_repository_bun
 from backend.features.alerts import AlertService
 from backend.features.review import TradeJournalService
 from backend.features.screener import StrategyTrackingService
+from backend.jobs import InProcessJobRunner, JobRegistry, SQLiteJobRunRepository
 from chart_data_provider import MarketChartClient
 from database import ReviewDatabase
 from ifind_client import IfindHttpClient
@@ -30,6 +31,7 @@ class ApplicationContainer:
     mentor_skills: MentorSkillRegistry
     realtime_aggregator: WebRealtimeAggregator
     chart_data: MarketChartClient
+    jobs: InProcessJobRunner
 
 
 def build_application_container(
@@ -41,6 +43,7 @@ def build_application_container(
 ) -> ApplicationContainer:
     data_gateway = build_data_gateway(credentials, tushare_token_supplier)
     repositories = build_repository_bundle(database)
+    jobs = InProcessJobRunner(JobRegistry.load(), SQLiteJobRunRepository(database))
     return ApplicationContainer(
         database=database,
         repositories=repositories,
@@ -53,4 +56,5 @@ def build_application_container(
         mentor_skills=MentorSkillRegistry(mentor_skills_dir, private_mentor_skills_dir),
         realtime_aggregator=data_gateway.realtime_observer,
         chart_data=data_gateway.chart_data,
+        jobs=jobs,
     )

backend/database/migrations/__init__.py

@@ -1,6 +1,7 @@
 from .m0001_adopt_legacy import MIGRATION as M0001_ADOPT_LEGACY
+from .m0002_job_runs import MIGRATION as M0002_JOB_RUNS
 from .runner import Migration, MigrationError, MigrationRunner
 
-MIGRATIONS = (M0001_ADOPT_LEGACY,)
+MIGRATIONS = (M0001_ADOPT_LEGACY, M0002_JOB_RUNS)
 
 __all__ = ["MIGRATIONS", "Migration", "MigrationError", "MigrationRunner"]

backend/database/migrations/m0002_job_runs.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import sqlite3
+
+from backend.database.migrations.runner import Migration
+
+
+def create_job_runs(connection: sqlite3.Connection) -> None:
+    connection.execute(
+        """
+        CREATE TABLE IF NOT EXISTS job_runs (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            job_id TEXT NOT NULL,
+            idempotency_key TEXT NOT NULL,
+            status TEXT NOT NULL,
+            attempt INTEGER NOT NULL DEFAULT 1,
+            started_at TEXT NOT NULL,
+            finished_at TEXT,
+            elapsed_ms INTEGER NOT NULL DEFAULT 0,
+            error_code TEXT NOT NULL DEFAULT '',
+            message TEXT NOT NULL DEFAULT '',
+            output_version TEXT NOT NULL DEFAULT '',
+            metadata TEXT NOT NULL DEFAULT '{}',
+            UNIQUE(job_id, idempotency_key, attempt)
+        )
+        """
+    )
+    connection.execute(
+        """
+        CREATE INDEX IF NOT EXISTS idx_job_runs_job_started
+        ON job_runs(job_id, started_at DESC, id DESC)
+        """
+    )
+    connection.execute(
+        """
+        CREATE INDEX IF NOT EXISTS idx_job_runs_status
+        ON job_runs(status, started_at DESC, id DESC)
+        """
+    )
+
+
+MIGRATION = Migration(
+    version="0002",
+    name="create_job_runs",
+    action=create_job_runs,
+    signature="job-runs:v1:id,job,key,status,attempt,times,elapsed,error,output,metadata",
+)

backend/jobs/__init__.py

@@ -0,0 +1,5 @@
+from .registry import JobDefinition, JobRegistry
+from .repository import SQLiteJobRunRepository
+from .runner import InProcessJobRunner
+
+__all__ = ["InProcessJobRunner", "JobDefinition", "JobRegistry", "SQLiteJobRunRepository"]

backend/jobs/registry.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+
+from app_config import APP_DIR
+
+
+@dataclass(frozen=True)
+class JobDefinition:
+    job_id: str
+    schedule: str
+    input_date_policy: str
+    dependencies: tuple[str, ...]
+    lock_key: str
+    timeout_seconds: int
+    max_attempts: int
+    output_version: str
+
+
+class JobRegistry:
+    def __init__(self, definitions: tuple[JobDefinition, ...]) -> None:
+        self.definitions = definitions
+        self._by_id = {item.job_id: item for item in definitions}
+        if len(self._by_id) != len(definitions):
+            raise ValueError("Background job IDs must be unique")
+
+    @classmethod
+    def load(cls, path: Path | None = None) -> "JobRegistry":
+        config_path = path or APP_DIR / "config" / "jobs.config.json"
+        payload = json.loads(config_path.read_text(encoding="utf-8"))
+        definitions = tuple(
+            JobDefinition(
+                job_id=str(item["id"]),
+                schedule=str(item["schedule"]),
+                input_date_policy=str(item["input_date_policy"]),
+                dependencies=tuple(str(value) for value in item.get("dependencies") or []),
+                lock_key=str(item["lock_key"]),
+                timeout_seconds=max(1, int(item["timeout_seconds"])),
+                max_attempts=max(1, int(item["max_attempts"])),
+                output_version=str(item["output_version"]),
+            )
+            for item in payload.get("jobs") or []
+        )
+        if not definitions:
+            raise ValueError("Background job registry is empty")
+        return cls(definitions)
+
+    def get(self, job_id: str) -> JobDefinition:
+        try:
+            return self._by_id[job_id]
+        except KeyError as exc:
+            raise ValueError(f"Background job is not registered: {job_id}") from exc

backend/jobs/repository.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from database import ReviewDatabase
+
+
+@dataclass(frozen=True)
+class SQLiteJobRunRepository:
+    database: ReviewDatabase
+
+    def completed(self, job_id: str, idempotency_key: str) -> bool:
+        with self.database.connect() as connection:
+            row = connection.execute(
+                """
+                SELECT 1 FROM job_runs
+                WHERE job_id = ? AND idempotency_key = ? AND status = 'success'
+                LIMIT 1
+                """,
+                (job_id, idempotency_key),
+            ).fetchone()
+        return row is not None
+
+    def start(
+        self, job_id: str, idempotency_key: str, output_version: str,
+        metadata: dict[str, Any] | None = None,
+    ) -> int:
+        now = datetime.now().astimezone().isoformat(timespec="seconds")
+        with self.database.connect() as connection:
+            row = connection.execute(
+                """
+                SELECT COALESCE(MAX(attempt), 0) + 1 AS attempt FROM job_runs
+                WHERE job_id = ? AND idempotency_key = ?
+                """,
+                (job_id, idempotency_key),
+            ).fetchone()
+            cursor = connection.execute(
+                """
+                INSERT INTO job_runs
+                    (job_id, idempotency_key, status, attempt, started_at,
+                     output_version, metadata)
+                VALUES (?, ?, 'running', ?, ?, ?, ?)
+                """,
+                (
+                    job_id, idempotency_key, int(row["attempt"]), now,
+                    output_version,
+                    json.dumps(metadata or {}, ensure_ascii=False, separators=(",", ":")),
+                ),
+            )
+            connection.execute(
+                """
+                DELETE FROM job_runs
+                WHERE id < (SELECT COALESCE(MAX(id), 0) - 20000 FROM job_runs)
+                  AND status != 'running'
+                """
+            )
+            return int(cursor.lastrowid)
+
+    def finish(
+        self, run_id: int, status: str, elapsed_ms: int,
+        error_code: str = "", message: str = "",
+    ) -> None:
+        now = datetime.now().astimezone().isoformat(timespec="seconds")
+        with self.database.connect() as connection:
+            connection.execute(
+                """
+                UPDATE job_runs
+                SET status = ?, finished_at = ?, elapsed_ms = ?,
+                    error_code = ?, message = ?
+                WHERE id = ?
+                """,
+                (status, now, elapsed_ms, error_code, message[:1000], int(run_id)),
+            )
+
+    def recent(self, limit: int = 20) -> list[dict[str, Any]]:
+        with self.database.connect() as connection:
+            rows = connection.execute(
+                """
+                SELECT id, job_id, idempotency_key, status, attempt, started_at,
+                       finished_at, elapsed_ms, error_code, message, output_version
+                FROM job_runs ORDER BY id DESC LIMIT ?
+                """,
+                (max(1, min(100, int(limit))),),
+            ).fetchall()
+        return [dict(row) for row in rows]

backend/jobs/runner.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import threading
+import time
+from collections.abc import Callable
+from typing import Any
+
+from backend.jobs.registry import JobRegistry
+from backend.jobs.repository import SQLiteJobRunRepository
+
+
+JobAction = Callable[[], Any]
+
+
+class InProcessJobRunner:
+    def __init__(self, registry: JobRegistry, repository: SQLiteJobRunRepository) -> None:
+        self.registry = registry
+        self.repository = repository
+        self._locks: dict[str, threading.Lock] = {}
+        self._locks_guard = threading.Lock()
+
+    def submit(
+        self, job_id: str, idempotency_key: str, action: JobAction,
+        metadata: dict[str, Any] | None = None,
+    ) -> bool:
+        definition = self.registry.get(job_id)
+        if self.repository.completed(job_id, idempotency_key):
+            return False
+        lock = self._lock(definition.lock_key)
+        if not lock.acquire(blocking=False):
+            return False
+        thread = threading.Thread(
+            target=self._execute_locked,
+            args=(job_id, idempotency_key, action, metadata, lock),
+            name=f"job-{job_id}-{idempotency_key}"[:80],
+            daemon=True,
+        )
+        thread.start()
+        return True
+
+    def run_inline(
+        self, job_id: str, idempotency_key: str, action: JobAction,
+        metadata: dict[str, Any] | None = None,
+    ) -> bool:
+        definition = self.registry.get(job_id)
+        if self.repository.completed(job_id, idempotency_key):
+            return False
+        lock = self._lock(definition.lock_key)
+        if not lock.acquire(blocking=False):
+            return False
+        self._execute_locked(job_id, idempotency_key, action, metadata, lock)
+        return True
+
+    def start_scheduler(
+        self, callback: Callable[[], None], stop_event: threading.Event,
+        interval_seconds: float, initial_delay_seconds: float = 0,
+    ) -> threading.Thread:
+        def schedule_loop() -> None:
+            if stop_event.wait(initial_delay_seconds):
+                return
+            while not stop_event.is_set():
+                try:
+                    callback()
+                except Exception:
+                    # Submitted jobs persist their own failures; the scheduler must stay alive.
+                    pass
+                stop_event.wait(interval_seconds)
+
+        thread = threading.Thread(
+            target=schedule_loop,
+            name="background-job-scheduler",
+            daemon=True,
+        )
+        thread.start()
+        return thread
+
+    def wait_for_idle(self, timeout_seconds: float = 5) -> bool:
+        deadline = time.monotonic() + max(0, timeout_seconds)
+        while time.monotonic() <= deadline:
+            with self._locks_guard:
+                busy = any(lock.locked() for lock in self._locks.values())
+            if not busy:
+                return True
+            time.sleep(0.01)
+        return False
+
+    def _execute_locked(
+        self, job_id: str, idempotency_key: str, action: JobAction,
+        metadata: dict[str, Any] | None, lock: threading.Lock,
+    ) -> None:
+        definition = self.registry.get(job_id)
+        try:
+            for attempt in range(1, definition.max_attempts + 1):
+                run_id = self.repository.start(
+                    job_id, idempotency_key, definition.output_version, metadata
+                )
+                started = time.perf_counter()
+                try:
+                    result = action()
+                    if isinstance(result, dict) and result.get("status") == "failed":
+                        raise RuntimeError(str(result.get("error") or "Job reported failure"))
+                    elapsed_ms = round((time.perf_counter() - started) * 1000)
+                    self.repository.finish(run_id, "success", elapsed_ms)
+                    return
+                except Exception as exc:
+                    elapsed_ms = round((time.perf_counter() - started) * 1000)
+                    self.repository.finish(
+                        run_id, "failed", elapsed_ms,
+                        type(exc).__name__, str(exc),
+                    )
+                    if attempt >= definition.max_attempts:
+                        return
+        finally:
+            lock.release()
+
+    def _lock(self, lock_key: str) -> threading.Lock:
+        with self._locks_guard:
+            return self._locks.setdefault(lock_key, threading.Lock())

config/README.md

@@ -11,6 +11,8 @@ These registries describe the approved product surface during architecture migra
   known blocked datasets.
 - `data-quality.config.json`: freshness, coverage, units, adjustment, point-in-time, and
   fail-closed rules for every canonical data product.
+- `jobs.config.json`: background schedules, dependencies, lock keys, retry policy, timeouts,
+  and output versions.
 
 During Stage 04 these files are contract inputs, not runtime replacements. Backend access in
 `api_access.py` remains authoritative until the HTTP governance phase switches it atomically.

config/jobs.config.json

@@ -0,0 +1,35 @@
+{
+  "schema_version": 1,
+  "jobs": [
+    {
+      "id": "market.refresh",
+      "schedule": "realtime polling or administrator request",
+      "input_date_policy": "requested trade date",
+      "dependencies": ["market provider", "database"],
+      "lock_key": "market-refresh",
+      "timeout_seconds": 120,
+      "max_attempts": 1,
+      "output_version": "dashboard-v1"
+    },
+    {
+      "id": "screener.automatic",
+      "schedule": "trading day after 15:10 Asia/Shanghai",
+      "input_date_policy": "current completed trade date",
+      "dependencies": ["market.refresh", "factor data", "database"],
+      "lock_key": "automatic-screener",
+      "timeout_seconds": 900,
+      "max_attempts": 1,
+      "output_version": "screener-library-v8"
+    },
+    {
+      "id": "market.ifind-event-enrichment",
+      "schedule": "on demand after market close",
+      "input_date_policy": "completed trade date",
+      "dependencies": ["ifind", "database"],
+      "lock_key": "ifind-event-enrichment",
+      "timeout_seconds": 180,
+      "max_attempts": 1,
+      "output_version": "ifind-event-v1"
+    }
+  ]
+}

docs/governance/architecture-inventory.json

@@ -189,8 +189,7 @@
     "heaven_readings"
   ],
   "background_job_methods": [
-    "_background_refresh_loop",
-    "_run_background_sync",
+    "_background_refresh_tick",
     "_schedule_automatic_screeners",
     "_schedule_ifind_event_enrichment",
     "run_automatic_screeners"
@@ -265,8 +264,8 @@
     },
     {
       "path": "server.py",
-      "bytes": 268422,
-      "lines": 5960
+      "bytes": 268447,
+      "lines": 5956
     },
     {
       "path": "static/redesign-v2.css",

docs/governance/stage-12-background-jobs.md

@@ -0,0 +1,30 @@
+# Stage 12: Governed Background Jobs
+
+Date: 2026-07-29
+
+## Result
+
+- Registered market refresh, automatic screening, and iFinD event enrichment as versioned
+  jobs with schedules, date policy, dependencies, locks, timeouts, retry limits, and output
+  versions.
+- Added the `job_runs` migration and persistent run ledger.
+- Centralized worker and scheduler thread creation in `InProcessJobRunner`.
+- Added process-level lock keys and persistent idempotency keys.
+- Persisted successful, failed, and retried attempts with elapsed time and normalized error
+  type.
+- Exposed recent job runs in administrator system status.
+- Limited retained completed history while preserving running jobs.
+- Kept existing feature services and the detailed market `sync_runs` audit unchanged.
+
+## Compatibility
+
+The runner remains in-process, matching the current single-container deployment. Jobs call the
+same application methods as HTTP requests and do not depend on request-local account state.
+The persistent contract permits a later worker process without changing job identities or
+business calculations.
+
+## Timeout Boundary
+
+Timeouts are declared and elapsed time is recorded. Python threads cannot be terminated safely,
+so hard cancellation remains cooperative until jobs move to a durable worker process. Locks and
+idempotency prevent concurrent duplicate execution in the current single-process deployment.

server.py

@@ -192,14 +192,15 @@ class DashboardService:
         self.mentor_skills = self.container.mentor_skills
         self.realtime_aggregator = self.container.realtime_aggregator
         self.chart_data = self.container.chart_data
+        self.jobs = self.container.jobs
         self.screener.ensure_builtin_strategies()
         self._background_stop = threading.Event()
-        self._background_thread = threading.Thread(
-            target=self._background_refresh_loop,
-            name="market-background-refresh",
-            daemon=True,
+        self._background_thread = self.jobs.start_scheduler(
+            self._background_refresh_tick,
+            self._background_stop,
+            interval_seconds=5,
+            initial_delay_seconds=3,
         )
-        self._background_thread.start()
 
     def _tushare_client(self) -> TushareClient:
         gateway = getattr(self, "data_gateway", None)
@@ -583,6 +584,7 @@ class DashboardService:
                     self._system_credentials.get("background_refresh_enabled", True)
                 ),
                 **self.database.status(),
+                "jobs": self.jobs.repository.recent(12),
             },
             "llm": {
                 "primary_configured": self._profile_configured(platform["primary"]),
@@ -772,36 +774,32 @@ class DashboardService:
             raise ValueError("用户不存在。")
 
     def request_background_sync(self, trade_date: str) -> bool:
-        if self.sync_lock.locked():
-            return False
         normalized = normalize_date(trade_date)
-        threading.Thread(
-            target=self._run_background_sync,
-            args=(normalized,),
-            name=f"market-sync-{normalized}",
-            daemon=True,
-        ).start()
-        return True
+        key = f"manual:{normalized}:{time.time_ns()}"
+        return self.jobs.submit(
+            "market.refresh",
+            key,
+            lambda: self.sync_dashboard(normalized),
+            {"trade_date": normalized, "trigger": "administrator"},
+        )
 
-    def _run_background_sync(self, trade_date: str) -> None:
-        try:
-            self.sync_dashboard(trade_date)
-        except Exception:
+    def _background_refresh_tick(self) -> None:
+        if not (
+            self.configured
+            and self._system_credentials.get("background_refresh_enabled", True)
+        ):
             return
-
-    def _background_refresh_loop(self) -> None:
-        self._background_stop.wait(3)
-        while not self._background_stop.is_set():
-            try:
-                if self.configured and self._system_credentials.get("background_refresh_enabled", True):
-                    today = date.today().strftime("%Y%m%d")
-                    snapshot = self.database.get_snapshot(today) or {}
-                    if self._realtime_snapshot_due(today, snapshot):
-                        self._run_background_sync(today)
-                    self._schedule_automatic_screeners(today, snapshot)
-            except Exception:
-                pass
-            self._background_stop.wait(5)
+        today = date.today().strftime("%Y%m%d")
+        snapshot = self.database.get_snapshot(today) or {}
+        if self._realtime_snapshot_due(today, snapshot):
+            bucket = int(time.time() // 5)
+            self.jobs.submit(
+                "market.refresh",
+                f"realtime:{today}:{bucket}",
+                lambda: self.sync_dashboard(today),
+                {"trade_date": today, "trigger": "realtime-poll"},
+            )
+        self._schedule_automatic_screeners(today, snapshot)
 
     def register_account(self, username: str, password: str) -> dict[str, Any]:
         username = username.strip()
@@ -1743,13 +1741,12 @@ class DashboardService:
         if last_attempt and (now - last_attempt).total_seconds() < 600:
             return False
         self._auto_screener_last_attempt[normalized_date] = now
-        threading.Thread(
-            target=self.run_automatic_screeners,
-            args=(normalized_date,),
-            name=f"automatic-screeners-{normalized_date}",
-            daemon=True,
-        ).start()
-        return True
+        return self.jobs.submit(
+            "screener.automatic",
+            f"{normalized_date}:v{SCREENER_LIBRARY_VERSION}",
+            lambda: self.run_automatic_screeners(normalized_date),
+            {"trade_date": normalized_date, "trigger": "post-close"},
+        )
 
     def run_automatic_screeners(self, trade_date: str) -> dict[str, Any]:
         normalized_date = normalize_date(trade_date)
@@ -4586,13 +4583,12 @@ class DashboardService:
         now = datetime.now().astimezone()
         if trade_date == now.strftime("%Y%m%d") and now.time().replace(tzinfo=None) < dt_time(15, 0):
             return
-        thread = threading.Thread(
-            target=self._refresh_ifind_event_enrichment,
-            args=(trade_date,),
-            name=f"ifind-event-{trade_date}",
-            daemon=True,
+        self.jobs.submit(
+            "market.ifind-event-enrichment",
+            f"{trade_date}:v1",
+            lambda: self._refresh_ifind_event_enrichment(trade_date),
+            {"trade_date": trade_date, "trigger": "dashboard-enrichment"},
         )
-        thread.start()
 
     def _refresh_ifind_event_enrichment(self, trade_date: str) -> None:
         if not self._ifind_event_lock.acquire(blocking=False):

tests/test_database_migrations.py

@@ -20,14 +20,17 @@ class DatabaseMigrationTests(unittest.TestCase):
                 ).fetchall()
             self.assertEqual(
                 [(row["version"], row["name"]) for row in rows],
-                [("0001", "adopt_legacy_schema")],
+                [
+                    ("0001", "adopt_legacy_schema"),
+                    ("0002", "create_job_runs"),
+                ],
             )
             ReviewDatabase(path)
             with database.connect() as connection:
                 count = connection.execute(
                     "SELECT COUNT(*) AS count FROM schema_migrations"
                 ).fetchone()["count"]
-            self.assertEqual(count, 1)
+            self.assertEqual(count, 2)
 
     def test_connection_factory_enables_required_pragmas(self) -> None:
         with tempfile.TemporaryDirectory() as root:

tests/test_governance_registries.py

@@ -23,6 +23,7 @@ class GovernanceRegistryTests(unittest.TestCase):
         self.api = load("api.config.json")
         self.data = load("data-fields.config.json")
         self.quality = load("data-quality.config.json")
+        self.jobs = load("jobs.config.json")
 
     def test_features_are_unique_and_use_declared_roles(self) -> None:
         roles = set(self.features["roles"])
@@ -92,6 +93,18 @@ class GovernanceRegistryTests(unittest.TestCase):
             all(str(rule.get("unit_profile") or "none") in profiles for rule in rules.values())
         )
 
+    def test_background_jobs_have_complete_unique_runtime_contracts(self) -> None:
+        jobs = self.jobs["jobs"]
+        ids = [item["id"] for item in jobs]
+        self.assertEqual(len(ids), len(set(ids)))
+        for item in jobs:
+            self.assertTrue(item["schedule"])
+            self.assertTrue(item["input_date_policy"])
+            self.assertTrue(item["lock_key"])
+            self.assertGreater(int(item["timeout_seconds"]), 0)
+            self.assertGreater(int(item["max_attempts"]), 0)
+            self.assertTrue(item["output_version"])
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_job_runner.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+from backend.jobs import InProcessJobRunner, JobRegistry, SQLiteJobRunRepository
+from database import ReviewDatabase
+
+
+class JobRunnerTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.database = ReviewDatabase(Path(self.temporary.name) / "review.db")
+        self.repository = SQLiteJobRunRepository(self.database)
+        self.runner = InProcessJobRunner(JobRegistry.load(), self.repository)
+
+    def test_successful_idempotent_job_runs_once(self) -> None:
+        calls = []
+        self.assertTrue(
+            self.runner.run_inline(
+                "screener.automatic", "20260729:v8", lambda: calls.append("run")
+            )
+        )
+        self.assertFalse(
+            self.runner.run_inline(
+                "screener.automatic", "20260729:v8", lambda: calls.append("again")
+            )
+        )
+        self.assertEqual(calls, ["run"])
+        self.assertEqual(self.repository.recent(1)[0]["status"], "success")
+
+    def test_failure_is_persisted_and_can_be_retried_later(self) -> None:
+        def fail() -> None:
+            raise RuntimeError("provider unavailable")
+
+        self.assertTrue(self.runner.run_inline("market.refresh", "failed-key", fail))
+        failed = self.repository.recent(1)[0]
+        self.assertEqual(failed["status"], "failed")
+        self.assertEqual(failed["error_code"], "RuntimeError")
+        self.assertTrue(
+            self.runner.run_inline("market.refresh", "failed-key", lambda: None)
+        )
+        self.assertEqual(self.repository.recent(1)[0]["attempt"], 2)
+
+    def test_lock_key_rejects_concurrent_submission(self) -> None:
+        entered = threading.Event()
+        release = threading.Event()
+
+        def wait() -> None:
+            entered.set()
+            release.wait(2)
+
+        self.assertTrue(self.runner.submit("market.refresh", "first", wait))
+        self.assertTrue(entered.wait(1))
+        self.assertFalse(self.runner.submit("market.refresh", "second", lambda: None))
+        release.set()
+        self.assertTrue(self.runner.wait_for_idle())
+
+    def test_failed_status_payload_is_recorded_as_failure(self) -> None:
+        self.assertTrue(
+            self.runner.run_inline(
+                "screener.automatic", "reported-failure",
+                lambda: {"status": "failed", "error": "missing factors"},
+            )
+        )
+        self.assertEqual(self.repository.recent(1)[0]["status"], "failed")
+
+
+if __name__ == "__main__":
+    unittest.main()