diff --git a/apps/api/src/main.py b/apps/api/src/main.py index 7e43b3e49..9e203f95e 100644 --- a/apps/api/src/main.py +++ b/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: diff --git a/apps/api/src/workers/signal_worker.py b/apps/api/src/workers/signal_worker.py index 1ef309923..cd82cb069 100644 --- a/apps/api/src/workers/signal_worker.py +++ b/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() diff --git a/apps/api/tests/test_runtime_bootstrap_guards.py b/apps/api/tests/test_runtime_bootstrap_guards.py index 16009b268..cdc406b33 100644 --- a/apps/api/tests/test_runtime_bootstrap_guards.py +++ b/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) + )