fix(automation): run asset reconciliation on worker

apps/api/tests/test_ai_automation_asset_capability_matrix.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
 
+import src.jobs.asset_capability_reconciliation_job as reconciliation_job
 import src.services.ai_automation_asset_capability_matrix as matrix_service
 from src.jobs.asset_capability_reconciliation_job import (
+    _persist_reconciliation,
     build_reconciliation_summary_payload,
     build_reconciliation_work_item_input,
 )
@@ -212,6 +215,7 @@ def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> N
             "candidate_count": 0,
             "repository_readback_count": 0,
             "repository_readback_verified": True,
+            "closed": True,
         },
     }
 
@@ -232,6 +236,38 @@ def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> N
     assert matrix["summary"]["gap_asset_count"] == 0
 
 
+def test_reconciliation_receipt_requires_closed_terminal() -> None:
+    now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC)
+    initial = build_asset_capability_matrix(
+        _single_host_scope(),
+        live_assets=[_fully_covered_host(now)],
+        latest_run={"run_id": "run-1", "status": "success", "ended_at": now},
+        repo_root=_REPO_ROOT,
+        generated_at=now,
+    )
+    matrix = build_asset_capability_matrix(
+        _single_host_scope(),
+        live_assets=[_fully_covered_host(now)],
+        latest_run={"run_id": "run-1", "status": "success", "ended_at": now},
+        reconciliation_receipt={
+            "status": "success",
+            "created_at": now,
+            "output": {
+                "matrix_fingerprint": initial["matrix_fingerprint"],
+                "candidate_count": 0,
+                "repository_readback_count": 0,
+                "repository_readback_verified": True,
+                "closed": False,
+            },
+        },
+        repo_root=_REPO_ROOT,
+        generated_at=now,
+    )
+
+    assert matrix["controls"]["daily_reconciliation_receipt"] is False
+    assert matrix["closed"] is False
+
+
 def test_outdated_live_receipts_are_classified_stale_per_stage() -> None:
     now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC)
     old = now - timedelta(hours=40)
@@ -342,6 +378,51 @@ def test_reconciliation_work_item_uses_same_trace_and_run_as_summary() -> None:
     )
 
 
+def test_reconciliation_persistence_batches_candidate_queries_and_inserts() -> None:
+    source = inspect.getsource(_persist_reconciliation)
+    loop_start = source.index("for candidate_raw in candidates:")
+    loop_end = source.index("if insert_rows:")
+
+    assert "ANY(CAST(:candidate_fingerprints AS text[]))" in source
+    assert "insert_rows" in source
+    assert "await db.execute" not in source[loop_start:loop_end]
+    assert "AND output ->> 'closed' = 'true'" in source
+
+
+def test_reconciliation_loop_retries_degraded_result_before_daily_wait(
+    monkeypatch: Any,
+) -> None:
+    triggered_by_values: list[str] = []
+    sleep_values: list[int] = []
+
+    async def _fake_reconcile_once(*, triggered_by: str) -> dict[str, object]:
+        triggered_by_values.append(triggered_by)
+        if len(triggered_by_values) == 1:
+            return {"status": "degraded_no_write", "reason": "transient_db_pressure"}
+        return {"status": "success", "closed": True}
+
+    async def _fake_sleep(seconds: int) -> None:
+        sleep_values.append(seconds)
+        if len(sleep_values) == 3:
+            raise asyncio.CancelledError
+
+    monkeypatch.setattr(reconciliation_job, "reconcile_once", _fake_reconcile_once)
+    monkeypatch.setattr(reconciliation_job.asyncio, "sleep", _fake_sleep)
+    monkeypatch.setattr(reconciliation_job, "_FIRST_DELAY_SECONDS", 90)
+    monkeypatch.setattr(reconciliation_job, "_LOOP_BACKOFF_SECONDS", 1_800)
+    monkeypatch.setattr(reconciliation_job, "_seconds_until_next_trigger", lambda: 1)
+
+    try:
+        asyncio.run(reconciliation_job.run_asset_capability_reconciliation_loop())
+    except asyncio.CancelledError:
+        pass
+    else:
+        raise AssertionError("loop should stop at the test cancellation point")
+
+    assert triggered_by_values == ["startup", "retry"]
+    assert sleep_values == [90, 1_800, 1]
+
+
 def test_priority_overlay_projects_aia_p0_006_runtime_controls() -> None:
     matrix = build_asset_capability_matrix(
         _single_host_scope(),

apps/api/tests/test_signal_worker_ansible_executor_binding.py

@@ -21,6 +21,7 @@ def test_signal_worker_produces_candidates_and_owns_security_maintenance() -> No
     assert "run_compliance_scanner_loop" in source
     assert "run_coverage_evaluator_loop" in source
     assert "run_asset_change_tracker_loop" in source
+    assert "run_asset_capability_reconciliation_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
@@ -33,6 +34,25 @@ def test_signal_worker_produces_candidates_and_owns_security_maintenance() -> No
     assert "signal_worker_ansible_check_mode_loop_started" not in source
 
 
+def test_signal_worker_is_single_asset_capability_reconciliation_owner() -> None:
+    from src import main as api_main
+
+    api_source = inspect.getsource(api_main.lifespan)
+    worker_source = inspect.getsource(signal_worker._main)
+
+    assert "run_asset_capability_reconciliation_loop" not in api_source
+    assert "asset_capability_reconciliation_owner_delegated" in api_source
+    assert "run_asset_capability_reconciliation_loop" in worker_source
+    maintenance_start = worker_source.index("security_maintenance_loops =")
+    maintenance_end = worker_source.index(
+        "security_maintenance_tasks =", maintenance_start
+    )
+    assert (
+        "run_asset_capability_reconciliation_loop"
+        in worker_source[maintenance_start:maintenance_end]
+    )
+
+
 def test_dedicated_broker_owns_ansible_executor_loop() -> None:
     source = inspect.getsource(ansible_executor_broker._main)
 
@@ -198,7 +218,10 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None:
     assert "legacy cluster-wide executor binding still exists" in workflow
     assert "awoooi-executor-boundary-verification" in workflow
     assert "awoooi_executor_boundary_verification_v4" in workflow
-    assert "single_writer_source_verifier=cd_tests_and_production_source_sha_v1" in workflow
+    assert (
+        "single_writer_source_verifier=cd_tests_and_production_source_sha_v1"
+        in workflow
+    )
     assert "decision_manager_queue_only=true" in workflow
     assert "webhook_router_queue_only=true" in workflow
     assert "failure_watcher_queue_only=true" in workflow
@@ -206,24 +229,27 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None:
     assert "candidate_idempotency_contract_verified=true" in workflow
     assert "apply_idempotency_contract_verified=true" in workflow
     assert "critical_break_glass_contract_verified=true" in workflow
-    assert "independent_post_verifier_source_verifier=cd_tests_and_production_source_sha_v1" in workflow
+    assert (
+        "independent_post_verifier_source_verifier=cd_tests_and_production_source_sha_v1"
+        in workflow
+    )
     assert "asset_postcondition_registry_complete=true" in workflow
     assert "independent_post_verifier_read_only=true" in workflow
     assert "executor_returncode_not_success_source=true" in workflow
     assert "verifier_failure_marks_apply_failed=true" in workflow
     assert "verifier_failure_retry_first=true" in workflow
     assert "post_verifier_raw_output_not_persisted=true" in workflow
-    assert "canonical_learning_source_verifier=cd_tests_and_production_source_sha_v1" in workflow
+    assert (
+        "canonical_learning_source_verifier=cd_tests_and_production_source_sha_v1"
+        in workflow
+    )
     assert "canonical_playbook_resolver_complete=true" in workflow
     assert "km_row_readback_required=true" in workflow
     assert "playbook_row_readback_required=true" in workflow
     assert "learning_failure_fail_closed=true" in workflow
     assert "trust_failure_fail_closed=true" in workflow
     assert "learning_receipts_public_safe=true" in workflow
-    assert (
-        "k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml"
-        in workflow
-    )
+    assert "k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml" in workflow
     assert "executor_boundary_public_readback=verified_ready" in workflow
     assert "EXECUTOR_BOUNDARY_READBACK_ATTEMPTS" in workflow
     assert "range(1, attempts + 1)" in workflow
@@ -234,9 +260,7 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None:
     assert "autonomous_single_writer_source_boundary" in workflow
     assert "independent_post_verifier_source_boundary" in workflow
     assert "canonical_learning_source_boundary" in workflow
-    worker_env = worker_deployment["spec"]["template"]["spec"]["containers"][0][
-        "env"
-    ]
+    worker_env = worker_deployment["spec"]["template"]["spec"]["containers"][0]["env"]
     assert any(item.get("name") == "AWOOOI_BUILD_COMMIT_SHA" for item in worker_env)
     assert "k8s/awoooi-prod/08-deployment-worker.yaml" in workflow
     assert '"source_boundary_verified"' in workflow
@@ -254,9 +278,18 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None:
     assert "current_user AS role_name" in workflow
     assert "hashlib.sha256" in workflow
     assert "representative_db_preflights_verified" in workflow
-    assert 'api_required_connection_budget="$API_DB_REQUIRED_CONNECTION_BUDGET"' in workflow
-    assert 'worker_required_connection_budget="$WORKER_DB_REQUIRED_CONNECTION_BUDGET"' in workflow
-    assert 'broker_required_connection_budget="$BROKER_DB_REQUIRED_CONNECTION_BUDGET"' in workflow
+    assert (
+        'api_required_connection_budget="$API_DB_REQUIRED_CONNECTION_BUDGET"'
+        in workflow
+    )
+    assert (
+        'worker_required_connection_budget="$WORKER_DB_REQUIRED_CONNECTION_BUDGET"'
+        in workflow
+    )
+    assert (
+        'broker_required_connection_budget="$BROKER_DB_REQUIRED_CONNECTION_BUDGET"'
+        in workflow
+    )
     assert "api_representative_connection_probe_count" in workflow
     assert "worker_representative_connection_probe_count" in workflow
     assert "broker_representative_connection_probe_count" in workflow
@@ -330,7 +363,5 @@ 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 (
-        'production_process_owner="awoooi-ansible-executor-broker"' in main_source
-    )
+    assert 'production_process_owner="awoooi-ansible-executor-broker"' in main_source
     assert "awooop_ansible_candidate_backfill_worker_schedule_evaluated" in main_source