fix(agent99): move dispatch reconciler to signal worker

apps/api/src/main.py

@@ -465,37 +465,6 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
             handoff="awoooi-worker deployment owns Redis stream consumption",
         )
 
-    # Agent99 durable dispatch outcome reconciler. The worker reads only the
-    # authenticated public-safe relay outcome and closes the same PG run after
-    # verifier, KM/PlayBook, Telegram, and incident receipts are durable.
-    if (
-        settings.AGENT99_SRE_ALERT_RELAY_URL
-        and settings.AGENT99_SRE_ALERT_RELAY_TOKEN
-    ):
-        try:
-            from src.jobs.agent99_controlled_dispatch_reconciler_job import (
-                run_agent99_controlled_dispatch_reconciler_loop,
-            )
-
-            scheduled_task = schedule_api_background_task(
-                run_agent99_controlled_dispatch_reconciler_loop()
-            )
-            logger.info(
-                "agent99_controlled_dispatch_reconciler_schedule_evaluated",
-                scheduled_in_process=scheduled_task is not None,
-                interval_seconds=15,
-            )
-        except Exception as exc:
-            logger.warning(
-                "agent99_controlled_dispatch_reconciler_schedule_failed",
-                error=str(exc),
-            )
-    else:
-        logger.info(
-            "agent99_controlled_dispatch_reconciler_skipped",
-            reason="authenticated_relay_not_configured",
-        )
-
     # BUG-005 修復 2026-04-11: 啟動時掃描 Redis 中所有 state=ready 但未送 Telegram 的 token
     # dedup TTL 10 分鐘過期後，ready decisions 就沒有補送機制 → 長期卡在 ready 無人審核
     try:

apps/api/src/workers/signal_worker.py

@@ -622,6 +622,7 @@ async def _main() -> None:
     ansible_candidate_backfill_task: asyncio.Task[Any] | None = None
     backup_restore_legacy_backfill_task: asyncio.Task[Any] | None = None
     security_maintenance_tasks: list[asyncio.Task[Any]] = []
+    agent99_controlled_dispatch_reconciler_task: asyncio.Task[Any] | None = None
     if settings.ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER:
         from src.jobs.awooop_ansible_candidate_backfill_job import (
             run_awooop_ansible_candidate_backfill_loop,
@@ -686,6 +687,41 @@ async def _main() -> None:
             task_names=[task.get_name() for task in security_maintenance_tasks],
             task_count=len(security_maintenance_tasks),
         )
+
+    # 2026-07-14 Codex v1.4: Agent99 reconciliation has one runtime owner.
+    # The API reserves its single DB connection for requests, so durable
+    # dispatch completion belongs to this standalone worker after the CD-owned
+    # schema migration / optional local bootstrap block above has completed.
+    if (
+        settings.AGENT99_SRE_ALERT_RELAY_URL.strip()
+        and settings.AGENT99_SRE_ALERT_RELAY_TOKEN.strip()
+    ):
+        from src.jobs.agent99_controlled_dispatch_reconciler_job import (
+            run_agent99_controlled_dispatch_reconciler_loop,
+        )
+
+        agent99_controlled_dispatch_reconciler_task = asyncio.create_task(
+            run_agent99_controlled_dispatch_reconciler_loop(),
+            name="run_agent99_controlled_dispatch_reconciler_loop",
+        )
+        logger.info(
+            "signal_worker_agent99_controlled_dispatch_reconciler_started",
+            owner="signal_worker",
+            interval_seconds=15,
+            authenticated_relay_configured=True,
+        )
+    else:
+        logger.warning(
+            "signal_worker_agent99_controlled_dispatch_reconciler_skipped_fail_closed",
+            owner="signal_worker",
+            reason="authenticated_relay_not_configured",
+            relay_url_configured=bool(
+                settings.AGENT99_SRE_ALERT_RELAY_URL.strip()
+            ),
+            relay_token_configured=bool(
+                settings.AGENT99_SRE_ALERT_RELAY_TOKEN.strip()
+            ),
+        )
     # Setup graceful shutdown
     shutdown_event = asyncio.Event()
 
@@ -727,6 +763,12 @@ async def _main() -> None:
         task.cancel()
     if security_maintenance_tasks:
         await asyncio.gather(*security_maintenance_tasks, return_exceptions=True)
+    if agent99_controlled_dispatch_reconciler_task is not None:
+        agent99_controlled_dispatch_reconciler_task.cancel()
+        try:
+            await agent99_controlled_dispatch_reconciler_task
+        except asyncio.CancelledError:
+            pass
     await worker.stop()
     await close_worker_redis_pool()  # 關閉 Worker 專屬連線
     await close_redis_pool()

apps/api/tests/test_runtime_bootstrap_guards.py

@@ -402,12 +402,80 @@ def test_agent99_retry_migration_matches_startup_bootstrap() -> None:
     )
 
 
-def test_api_runs_retry_schema_bootstrap_before_agent99_reconciler() -> None:
+def test_signal_worker_owns_agent99_reconciler_after_schema_bootstrap_gate() -> None:
     import inspect
 
     from src import main as api_main
+    from src.workers import signal_worker
 
-    source = inspect.getsource(api_main.lifespan)
-    assert source.index("await init_db()") < source.index(
+    api_source = inspect.getsource(api_main.lifespan)
+    worker_source = inspect.getsource(signal_worker._main)
+
+    assert "run_agent99_controlled_dispatch_reconciler_loop" not in api_source
+    assert worker_source.count(
+        "run_agent99_controlled_dispatch_reconciler_loop()"
+    ) == 1
+    assert worker_source.index(
+        "if settings.DATABASE_BOOTSTRAP_DDL_ENABLED"
+    ) < worker_source.index(
         "run_agent99_controlled_dispatch_reconciler_loop()"
     )
+    assert 'owner="cd_migration_preflight"' in worker_source
+    assert 'owner="signal_worker"' in worker_source
+
+
+def test_signal_worker_agent99_reconciler_requires_url_and_token() -> None:
+    import inspect
+
+    from src.workers import signal_worker
+
+    source = inspect.getsource(signal_worker._main)
+    start = source.index("settings.AGENT99_SRE_ALERT_RELAY_URL.strip()")
+    stop = source.index("# Setup graceful shutdown")
+    ownership_block = source[start:stop]
+
+    assert "settings.AGENT99_SRE_ALERT_RELAY_TOKEN.strip()" in ownership_block
+    assert "authenticated_relay_not_configured" in ownership_block
+    assert (
+        "signal_worker_agent99_controlled_dispatch_reconciler_skipped_fail_closed"
+        in ownership_block
+    )
+
+
+def test_signal_worker_gracefully_cancels_agent99_reconciler() -> None:
+    import inspect
+
+    from src.workers import signal_worker
+
+    source = inspect.getsource(signal_worker._main)
+    cancel = "agent99_controlled_dispatch_reconciler_task.cancel()"
+    await_task = "await agent99_controlled_dispatch_reconciler_task"
+
+    assert cancel in source
+    assert await_task in source
+    assert source.index(cancel) < source.index(await_task)
+    assert source.index(await_task) < source.index("await worker.stop()")
+
+
+def test_production_signal_worker_is_single_agent99_reconciler_owner() -> None:
+    import inspect
+
+    from src import main as api_main
+    from src.workers import signal_worker
+
+    root = Path(__file__).resolve().parents[3]
+    manifest = yaml.safe_load(
+        (root / "k8s/awoooi-prod/08-deployment-worker.yaml").read_text(
+            encoding="utf-8"
+        )
+    )
+    container = manifest["spec"]["template"]["spec"]["containers"][0]
+
+    assert manifest["spec"]["replicas"] == 1
+    assert container["command"] == ["python", "-m", "src.workers.signal_worker"]
+    assert "run_agent99_controlled_dispatch_reconciler_loop" not in (
+        inspect.getsource(api_main.lifespan)
+    )
+    assert "run_agent99_controlled_dispatch_reconciler_loop" in (
+        inspect.getsource(signal_worker._main)
+    )