fix(security): restore control plane maintenance ownership

apps/api/src/core/config.py

@@ -751,6 +751,14 @@ class Settings(BaseSettings):
             "then enqueue them for the existing check-mode worker."
         ),
     )
+    ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER: bool = Field(
+        default=False,
+        description=(
+            "True=run externally read-only discovery plus bounded rule catalog, "
+            "compliance, coverage, and asset drift control-plane writebacks in the "
+            "dedicated signal worker."
+        ),
+    )
     AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS: int = Field(
         default=600,
         ge=60,

apps/api/src/workers/signal_worker.py

@@ -621,6 +621,7 @@ async def _main() -> None:
     worker = await init_signal_worker()
     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]] = []
     if settings.ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER:
         from src.jobs.awooop_ansible_candidate_backfill_job import (
             run_awooop_ansible_candidate_backfill_loop,
@@ -662,6 +663,29 @@ async def _main() -> None:
                 endpoint_class=relay_preflight["endpoint_class"],
                 telegram_dispatch_performed=False,
             )
+    if settings.ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER:
+        from src.jobs.asset_change_tracker_job import run_asset_change_tracker_loop
+        from src.jobs.asset_scanner_job import run_asset_scanner_loop
+        from src.jobs.compliance_scanner_job import run_compliance_scanner_loop
+        from src.jobs.coverage_evaluator_job import run_coverage_evaluator_loop
+        from src.jobs.rule_catalog_sync_job import run_rule_catalog_sync_loop
+
+        security_maintenance_loops = (
+            ("run_asset_scanner_loop", run_asset_scanner_loop),
+            ("run_rule_catalog_sync_loop", run_rule_catalog_sync_loop),
+            ("run_compliance_scanner_loop", run_compliance_scanner_loop),
+            ("run_coverage_evaluator_loop", run_coverage_evaluator_loop),
+            ("run_asset_change_tracker_loop", run_asset_change_tracker_loop),
+        )
+        security_maintenance_tasks = [
+            asyncio.create_task(loop(), name=task_name)
+            for task_name, loop in security_maintenance_loops
+        ]
+        logger.info(
+            "signal_worker_security_control_plane_maintenance_started",
+            task_names=[task.get_name() for task in security_maintenance_tasks],
+            task_count=len(security_maintenance_tasks),
+        )
     # Setup graceful shutdown
     shutdown_event = asyncio.Event()
 
@@ -699,6 +723,10 @@ async def _main() -> None:
             await backup_restore_legacy_backfill_task
         except asyncio.CancelledError:
             pass
+    for task in security_maintenance_tasks:
+        task.cancel()
+    if security_maintenance_tasks:
+        await asyncio.gather(*security_maintenance_tasks, return_exceptions=True)
     await worker.stop()
     await close_worker_redis_pool()  # 關閉 Worker 專屬連線
     await close_redis_pool()

apps/api/tests/test_signal_worker_ansible_executor_binding.py

@@ -8,13 +8,22 @@ import yaml
 from src.workers import ansible_executor_broker, signal_worker
 
 
-def test_signal_worker_only_produces_ansible_candidates() -> None:
+def test_signal_worker_produces_candidates_and_owns_security_maintenance() -> None:
     source = inspect.getsource(signal_worker._main)
 
     assert "run_awooop_ansible_candidate_backfill_loop" in source
     assert "signal_worker_ansible_candidate_backfill_loop_started" in source
     assert "ansible_candidate_backfill_task.cancel()" in source
     assert "await ansible_candidate_backfill_task" in source
+    assert "ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER" in source
+    assert "run_asset_scanner_loop" in source
+    assert "run_rule_catalog_sync_loop" in source
+    assert "run_compliance_scanner_loop" in source
+    assert "run_coverage_evaluator_loop" in source
+    assert "run_asset_change_tracker_loop" in source
+    assert "signal_worker_security_control_plane_maintenance_started" in source
+    assert "task.cancel()" in source
+    assert "await asyncio.gather(*security_maintenance_tasks" in source
     assert "run_awooop_ansible_check_mode_loop" not in source
     assert "signal_worker_ansible_check_mode_loop_started" not in source
 
@@ -37,6 +46,7 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None:
     assert 'command: ["python", "-m", "src.workers.signal_worker"]' in manifest
     assert "serviceAccountName: awoooi-worker" in manifest
     assert "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER" in manifest
+    assert "ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER" in manifest
     assert "AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS" in manifest
     assert "ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER" in manifest
     assert 'value: "false"' in manifest
@@ -45,6 +55,12 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None:
     assert "mountPath: /run/secrets/ssh_mcp_key" not in manifest
     assert "mountPath: /etc/ssh-mcp/known_hosts" not in manifest
     assert "secretName: ssh-mcp-key" not in manifest
+    deployment = yaml.safe_load(manifest)
+    worker_env = {
+        item["name"]: item.get("value")
+        for item in deployment["spec"]["template"]["spec"]["containers"][0]["env"]
+    }
+    assert worker_env["ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER"] == "true"
 
 
 def test_execution_broker_is_only_workload_with_ssh_transport() -> None:
@@ -95,6 +111,7 @@ def test_api_manifest_has_read_only_identity_and_no_ssh_executor_mounts() -> Non
     assert pod_spec["serviceAccountName"] == "awoooi-api"
     assert env["SSH_MCP_ENABLED"] == "false"
     assert env["DATABASE_NULL_POOL"] == "true"
+    assert env["AWOOOI_API_BACKGROUND_TASKS_ENABLED"] == "false"
     assert env["ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER"] == "false"
     assert env["ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER"] == "false"
     assert env["ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR"] == "true"

k8s/awoooi-prod/06-deployment-api.yaml

@@ -174,6 +174,8 @@ spec:
               # 2026-07-11 Codex: release each API connection after use while
               # workload-specific DB identities are being split and verified.
               value: "true"
+            - name: AWOOOI_API_BACKGROUND_TASKS_ENABLED
+              value: "false"
             - name: IWOOOS_WAZUH_READONLY_ENABLED
               # 2026-06-30 Codex: controlled GitOps enablement after owner
               # metadata, manager registry acceptance, dry-run, rollback, and

k8s/awoooi-prod/08-deployment-worker.yaml

@@ -176,6 +176,8 @@ spec:
               value: "968bd794d16f53192484a381477aa7c091ed9648"
             - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER
               value: "true"
+            - name: ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER
+              value: "true"
             - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS
               value: "600"
             - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT