fix(agent): co-locate ansible candidate producer

apps/api/src/main.py

@@ -638,10 +638,14 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
         from src.jobs.awooop_ansible_candidate_backfill_job import (
             run_awooop_ansible_candidate_backfill_loop,
         )
-        schedule_api_background_task(run_awooop_ansible_candidate_backfill_loop())
+        scheduled_task = schedule_api_background_task(
+            run_awooop_ansible_candidate_backfill_loop()
+        )
         logger.info(
-            "awooop_ansible_candidate_backfill_worker_scheduled",
+            "awooop_ansible_candidate_backfill_worker_schedule_evaluated",
             enabled=settings.ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER,
+            scheduled_in_api_process=scheduled_task is not None,
+            production_process_owner="awoooi-worker",
             interval_seconds=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS,
         )
     except Exception as e:

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_candidate_backfill_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,
+        )
+
+        ansible_candidate_backfill_task = asyncio.create_task(
+            run_awooop_ansible_candidate_backfill_loop(),
+            name="run_awooop_ansible_candidate_backfill_loop",
+        )
+        logger.info(
+            "signal_worker_ansible_candidate_backfill_loop_started",
+            interval_seconds=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS,
+            batch_limit=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT,
+        )
     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 (
@@ -649,6 +664,12 @@ async def _main() -> None:
 
     # Graceful shutdown
     logger.info("signal_worker_shutting_down")
+    if ansible_candidate_backfill_task is not None:
+        ansible_candidate_backfill_task.cancel()
+        try:
+            await ansible_candidate_backfill_task
+        except asyncio.CancelledError:
+            pass
     if ansible_check_mode_task is not None:
         ansible_check_mode_task.cancel()
         try:

apps/api/tests/test_signal_worker_ansible_executor_binding.py

@@ -9,6 +9,10 @@ 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_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 "run_awooop_ansible_check_mode_loop" in source
     assert "asyncio.create_task" in source
     assert "signal_worker_ansible_check_mode_loop_started" in source
@@ -23,6 +27,8 @@ def test_worker_manifest_mounts_existing_ssh_mcp_transport() -> None:
     assert 'command: ["python", "-m", "src.workers.signal_worker"]' in manifest
     assert "serviceAccountName: awoooi-executor" in manifest
     assert "fsGroup: 1000" in manifest
+    assert "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER" in manifest
+    assert "AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS" 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
@@ -39,3 +45,4 @@ def test_api_log_does_not_claim_skipped_executor_was_scheduled() -> None:
     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
+    assert "awooop_ansible_candidate_backfill_worker_schedule_evaluated" in main_source

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

@@ -93,6 +93,16 @@ spec:
               value: "1"
             - name: DATABASE_MAX_OVERFLOW
               value: "0"
+            - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER
+              value: "true"
+            - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS
+              value: "600"
+            - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT
+              value: "2"
+            - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS
+              value: "24"
+            - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS
+              value: "10"
             - name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER
               value: "true"
             - name: ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY