fix(agent): run Ansible executor in worker

apps/api/src/main.py

@@ -654,10 +654,14 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
         from src.jobs.awooop_ansible_check_mode_job import (
             run_awooop_ansible_check_mode_loop,
         )
-        schedule_api_background_task(run_awooop_ansible_check_mode_loop())
+        scheduled_task = schedule_api_background_task(
+            run_awooop_ansible_check_mode_loop()
+        )
         logger.info(
-            "awooop_ansible_check_mode_worker_scheduled",
+            "awooop_ansible_check_mode_worker_schedule_evaluated",
             enabled=settings.ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER,
+            scheduled_in_api_process=scheduled_task is not None,
+            production_process_owner="awoooi-worker",
             interval_seconds=settings.AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS,
         )
     except Exception as e:

apps/api/src/services/awooop_ansible_check_mode_service.py

@@ -390,6 +390,10 @@ async def _run_ansible_command(spec: AnsibleCommandSpec, *, timeout_seconds: int
         timed_out = True
         process.kill()
         stdout_bytes, stderr_bytes = await process.communicate()
+    except asyncio.CancelledError:
+        process.kill()
+        await process.communicate()
+        raise
     duration_ms = int((time.monotonic() - started) * 1000)
     return AnsibleRunResult(
         returncode=124 if timed_out else int(process.returncode or 0),

apps/api/src/workers/signal_worker.py

@@ -608,6 +608,21 @@ async def _main() -> None:
 
     # Initialize and start worker
     worker = await init_signal_worker()
+    ansible_check_mode_task: asyncio.Task[Any] | None = None
+    if settings.ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER:
+        from src.jobs.awooop_ansible_check_mode_job import (
+            run_awooop_ansible_check_mode_loop,
+        )
+
+        ansible_check_mode_task = asyncio.create_task(
+            run_awooop_ansible_check_mode_loop(),
+            name="run_awooop_ansible_check_mode_loop",
+        )
+        logger.info(
+            "signal_worker_ansible_check_mode_loop_started",
+            interval_seconds=settings.AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS,
+            batch_limit=settings.AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT,
+        )
 
     # Setup graceful shutdown
     shutdown_event = asyncio.Event()
@@ -634,6 +649,12 @@ async def _main() -> None:
 
     # Graceful shutdown
     logger.info("signal_worker_shutting_down")
+    if ansible_check_mode_task is not None:
+        ansible_check_mode_task.cancel()
+        try:
+            await ansible_check_mode_task
+        except asyncio.CancelledError:
+            pass
     await worker.stop()
     await close_worker_redis_pool()  # 關閉 Worker 專屬連線
     await close_redis_pool()

apps/api/tests/test_awooop_truth_chain_service.py

@@ -23,6 +23,7 @@ from src.services.awooop_ansible_check_mode_service import (
     _post_apply_verification_result,
     _record_auto_repair_execution_receipt,
     _record_learning_writeback_receipt,
+    _run_ansible_command,
     _send_controlled_apply_telegram_receipt,
     backfill_missing_auto_repair_execution_receipts_once,
     build_ansible_apply_command,
@@ -1785,6 +1786,14 @@ def test_ansible_check_and_apply_rows_persist_canonical_automation_run_id() -> N
     assert '"automation_run_id": str(' in apply_source
 
 
+def test_ansible_subprocess_is_terminated_when_worker_task_is_cancelled() -> None:
+    source = inspect.getsource(_run_ansible_command)
+
+    assert "except asyncio.CancelledError" in source
+    assert "process.kill()" in source
+    assert "await process.communicate()" in source
+
+
 def test_ansible_post_apply_verifier_helpers_are_deterministic() -> None:
     assert _post_apply_km_path_type("03ca6836-1b76-4da2-8e3e-6d3b6df9254a") == (
         "ansible_apply_receipt:03ca6836"

apps/api/tests/test_signal_worker_ansible_executor_binding.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import inspect
+from pathlib import Path
+
+from src.workers import signal_worker
+
+
+def test_signal_worker_owns_ansible_executor_loop() -> None:
+    source = inspect.getsource(signal_worker._main)
+
+    assert "run_awooop_ansible_check_mode_loop" in source
+    assert "asyncio.create_task" in source
+    assert "signal_worker_ansible_check_mode_loop_started" in source
+    assert "ansible_check_mode_task.cancel()" in source
+    assert "await ansible_check_mode_task" in source
+
+
+def test_worker_manifest_mounts_existing_ssh_mcp_transport() -> None:
+    repo_root = Path(__file__).resolve().parents[3]
+    manifest = (repo_root / "k8s/awoooi-prod/08-deployment-worker.yaml").read_text()
+
+    assert 'command: ["python", "-m", "src.workers.signal_worker"]' in manifest
+    assert "serviceAccountName: awoooi-executor" in manifest
+    assert "ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER" in manifest
+    assert "ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY" in manifest
+    assert "AWOOOP_ANSIBLE_CHECK_MODE_STARTUP_SLEEP_SECONDS" in manifest
+    assert "mountPath: /run/secrets/ssh_mcp_key" in manifest
+    assert "mountPath: /etc/ssh-mcp/known_hosts" in manifest
+    assert "secretName: ssh-mcp-key" in manifest
+    assert "optional: true" in manifest
+
+
+def test_api_log_does_not_claim_skipped_executor_was_scheduled() -> None:
+    repo_root = Path(__file__).resolve().parents[3]
+    main_source = (repo_root / "apps/api/src/main.py").read_text()
+
+    assert "awooop_ansible_check_mode_worker_schedule_evaluated" in main_source
+    assert "scheduled_in_api_process=scheduled_task is not None" in main_source
+    assert 'production_process_owner="awoooi-worker"' in main_source