fix(automation): stabilize asset reconciliation source

apps/api/tests/test_ai_automation_asset_capability_matrix.py

@@ -454,13 +454,24 @@ def test_reconcile_once_publishes_closed_public_matrix_cache(
     monkeypatch: Any,
 ) -> None:
     live_readback_calls: list[dict[str, object]] = []
+    settled_at = (datetime.now(UTC) - timedelta(minutes=10)).isoformat()
 
     async def _fake_live_readback(**kwargs: object) -> dict[str, object]:
         live_readback_calls.append(dict(kwargs))
         if kwargs["include_live_only_assets"] is True:
-            return {"live_readback_status": "ready"}
+            return {
+                "live_readback_status": "ready",
+                "matrix_fingerprint": "stable-fingerprint",
+                "latest_discovery_run": {
+                    "run_id": "discovery-run-1",
+                    "ended_at": settled_at,
+                },
+            }
         return {"live_readback_status": "ready", "closed": True}
 
+    async def _no_wait(_seconds: int) -> None:
+        return None
+
     async def _fake_persist(
         _matrix: object,
         _candidates: object,
@@ -481,6 +492,7 @@ def test_reconcile_once_publishes_closed_public_matrix_cache(
         lambda _matrix: [],
     )
     monkeypatch.setattr(reconciliation_job, "_persist_reconciliation", _fake_persist)
+    monkeypatch.setattr(reconciliation_job.asyncio, "sleep", _no_wait)
 
     result = asyncio.run(
         reconciliation_job.reconcile_once(
@@ -495,6 +507,11 @@ def test_reconcile_once_publishes_closed_public_matrix_cache(
             "timeout_seconds": 30.0,
             "include_live_only_assets": True,
         },
+        {
+            "project_id": "awoooi",
+            "timeout_seconds": 30.0,
+            "include_live_only_assets": True,
+        },
         {
             "project_id": "awoooi",
             "timeout_seconds": 30.0,
@@ -505,6 +522,95 @@ def test_reconcile_once_publishes_closed_public_matrix_cache(
     assert result["public_cache_refreshed"] is True
 
 
+def test_reconcile_once_waits_for_settled_discovery_before_write(
+    monkeypatch: Any,
+) -> None:
+    async def _fake_live_readback(**_kwargs: object) -> dict[str, object]:
+        return {
+            "live_readback_status": "ready",
+            "matrix_fingerprint": "too-fresh",
+            "latest_discovery_run": {
+                "run_id": "discovery-run-1",
+                "ended_at": datetime.now(UTC).isoformat(),
+            },
+        }
+
+    async def _unexpected_persist(*_args: object, **_kwargs: object) -> None:
+        raise AssertionError("unsettled source must not persist reconciliation")
+
+    monkeypatch.setattr(
+        reconciliation_job,
+        "build_asset_capability_matrix_with_live_readback",
+        _fake_live_readback,
+    )
+    monkeypatch.setattr(
+        reconciliation_job,
+        "_persist_reconciliation",
+        _unexpected_persist,
+    )
+
+    result = asyncio.run(
+        reconciliation_job.reconcile_once(
+            triggered_by="controlled_replay",
+            project_id="awoooi",
+        )
+    )
+
+    assert result["status"] == "source_snapshot_settling"
+    assert result["reason"] == "latest_discovery_run_not_settled"
+    assert result["retry_seconds"] >= 60
+    assert result["external_runtime_write_performed"] is False
+
+
+def test_reconcile_once_retries_when_source_changes_during_stability_check(
+    monkeypatch: Any,
+) -> None:
+    fingerprints = iter(("first-fingerprint", "second-fingerprint"))
+    settled_at = (datetime.now(UTC) - timedelta(minutes=10)).isoformat()
+
+    async def _fake_live_readback(**_kwargs: object) -> dict[str, object]:
+        return {
+            "live_readback_status": "ready",
+            "matrix_fingerprint": next(fingerprints),
+            "latest_discovery_run": {
+                "run_id": "discovery-run-1",
+                "ended_at": settled_at,
+            },
+        }
+
+    async def _no_wait(_seconds: int) -> None:
+        return None
+
+    async def _unexpected_persist(*_args: object, **_kwargs: object) -> None:
+        raise AssertionError("unstable source must not persist reconciliation")
+
+    monkeypatch.setattr(
+        reconciliation_job,
+        "build_asset_capability_matrix_with_live_readback",
+        _fake_live_readback,
+    )
+    monkeypatch.setattr(reconciliation_job.asyncio, "sleep", _no_wait)
+    monkeypatch.setattr(
+        reconciliation_job,
+        "_persist_reconciliation",
+        _unexpected_persist,
+    )
+
+    result = asyncio.run(
+        reconciliation_job.reconcile_once(
+            triggered_by="controlled_replay",
+            project_id="awoooi",
+        )
+    )
+
+    assert result == {
+        "status": "source_snapshot_unstable",
+        "reason": "matrix_changed_during_stability_verifier",
+        "retry_seconds": 60,
+        "external_runtime_write_performed": False,
+    }
+
+
 def test_reconciliation_loop_retries_degraded_result_before_poll(
     monkeypatch: Any,
 ) -> None:
@@ -524,9 +630,9 @@ def test_reconciliation_loop_retries_degraded_result_before_poll(
 
     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, "_FIRST_DELAY_SECONDS", 420)
     monkeypatch.setattr(reconciliation_job, "_LOOP_BACKOFF_SECONDS", 1_800)
-    monkeypatch.setattr(reconciliation_job, "_RECONCILIATION_POLL_SECONDS", 1)
+    monkeypatch.setattr(reconciliation_job, "_RECONCILIATION_POLL_SECONDS", 3_600)
 
     try:
         asyncio.run(reconciliation_job.run_asset_capability_reconciliation_loop())
@@ -536,7 +642,7 @@ def test_reconciliation_loop_retries_degraded_result_before_poll(
         raise AssertionError("loop should stop at the test cancellation point")
 
     assert triggered_by_values == ["startup", "retry"]
-    assert sleep_values == [90, 1_800, 1]
+    assert sleep_values == [420, 1_800, 3_600]
 
 
 def test_priority_overlay_projects_aia_p0_006_runtime_controls() -> None:

apps/api/tests/test_asset_change_tracker_job.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import src.jobs.asset_change_tracker_job as tracker
+
+
+class _FakeResult:
+    def __init__(self, rows: list[tuple[str]] | None = None) -> None:
+        self._rows = rows or []
+        self.rowcount = 0
+
+    def fetchall(self) -> list[tuple[str]]:
+        return self._rows
+
+
+class _FakeDb:
+    async def execute(
+        self,
+        statement: Any,
+        _params: dict[str, Any] | None = None,
+    ) -> _FakeResult:
+        if "SELECT run_id FROM asset_discovery_run" in str(statement):
+            return _FakeResult([("newer-run",), ("older-run",)])
+        return _FakeResult()
+
+
+class _FakeDbContext:
+    async def __aenter__(self) -> _FakeDb:
+        return _FakeDb()
+
+    async def __aexit__(self, *_args: object) -> None:
+        return None
+
+
+def test_change_tracker_scopes_every_database_path_to_awoooi(monkeypatch: Any) -> None:
+    project_ids: list[str] = []
+
+    def _get_db_context(project_id: str) -> _FakeDbContext:
+        project_ids.append(project_id)
+        return _FakeDbContext()
+
+    monkeypatch.setattr("src.db.base.get_db_context", _get_db_context)
+
+    assert asyncio.run(tracker._get_recent_runs()) == ["newer-run", "older-run"]
+    assert asyncio.run(tracker._diff_runs("newer-run", "older-run")) == {
+        "added": 0,
+        "removed": 0,
+        "modified": 0,
+    }
+    asyncio.run(tracker._log_aol({"added": 0, "removed": 0, "modified": 0}, 1, None))
+
+    assert project_ids == ["awoooi", "awoooi", "awoooi"]

apps/api/tests/test_asset_scanner_job.py

@@ -4,6 +4,8 @@ import asyncio
 import json
 from pathlib import Path
 
+import pytest
+
 from src.jobs import asset_scanner_job
 
 
@@ -1547,6 +1549,10 @@ def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) -
         "RuntimeError: asset_collector_failures=prometheus_targets"
     )
 
+    with pytest.raises(RuntimeError, match="asset_scan_failed"):
+        asyncio.run(asset_scanner_job.scan_once(raise_on_failure=True))
+    assert terminal["status"] == "failed"
+
 
 def test_database_catalog_collector_reads_metadata_only(monkeypatch) -> None:
     requested_projects: list[str | None] = []

apps/api/tests/test_rule_catalog_sync_job.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import src.jobs.rule_catalog_sync_job as rule_sync
+
+
+class _FakeResult:
+    def one_or_none(self) -> tuple[str, bool]:
+        return "rule-id", True
+
+
+class _FakeDb:
+    async def execute(
+        self,
+        _statement: Any,
+        _parameters: dict[str, Any] | None = None,
+    ) -> _FakeResult:
+        return _FakeResult()
+
+
+class _FakeDbContext:
+    async def __aenter__(self) -> _FakeDb:
+        return _FakeDb()
+
+    async def __aexit__(self, *_args: object) -> None:
+        return None
+
+
+def _rule(name: str = "ExampleRule") -> dict[str, Any]:
+    return {
+        "rule_name": name,
+        "expr": "vector(1)",
+        "duration_seconds": 60,
+        "severity": "warning",
+        "labels": {},
+        "annotations": {},
+        "group_name": "example",
+    }
+
+
+def test_rule_catalog_database_paths_are_tenant_scoped(monkeypatch: Any) -> None:
+    project_ids: list[str] = []
+
+    def _get_db_context(project_id: str) -> _FakeDbContext:
+        project_ids.append(project_id)
+        return _FakeDbContext()
+
+    monkeypatch.setattr("src.db.base.get_db_context", _get_db_context)
+
+    assert asyncio.run(rule_sync._upsert_rule(_rule())) == "new"
+    asyncio.run(
+        rule_sync._log_aol(
+            {"total": 1, "new": 1, "updated": 0, "unchanged": 0, "failed": 0},
+            duration_ms=1,
+            error=None,
+        )
+    )
+
+    assert project_ids == ["awoooi", "awoooi"]
+
+
+def test_rule_catalog_failure_is_visible_and_durably_logged(monkeypatch: Any) -> None:
+    captured: dict[str, Any] = {}
+
+    async def _fetch() -> list[dict[str, Any]]:
+        return [_rule("RuleA"), _rule("RuleB")]
+
+    async def _fail_upsert(_item: dict[str, Any]) -> str:
+        raise RuntimeError("database unavailable")
+
+    async def _capture_log(
+        stats: dict[str, int],
+        duration_ms: int,
+        error: str | None,
+    ) -> None:
+        captured.update(stats=stats.copy(), duration_ms=duration_ms, error=error)
+
+    monkeypatch.setattr(rule_sync, "_fetch_prometheus_rules", _fetch)
+    monkeypatch.setattr(rule_sync, "_upsert_rule", _fail_upsert)
+    monkeypatch.setattr(rule_sync, "_log_aol", _capture_log)
+
+    result = asyncio.run(rule_sync.sync_once())
+
+    assert result == {
+        "total": 2,
+        "new": 0,
+        "updated": 0,
+        "unchanged": 0,
+        "failed": 2,
+    }
+    assert str(captured["error"]).startswith("RuntimeError: database unavailable")
+    assert captured["stats"] == result
+
+
+def test_rule_catalog_loop_retries_failed_sync_before_hourly_wait(
+    monkeypatch: Any,
+) -> None:
+    sync_count = 0
+    sleep_values: list[int] = []
+
+    async def _sync_once() -> dict[str, int]:
+        nonlocal sync_count
+        sync_count += 1
+        return {
+            "total": 1,
+            "new": 0,
+            "updated": 0,
+            "unchanged": 0 if sync_count == 1 else 1,
+            "failed": 1 if sync_count == 1 else 0,
+        }
+
+    async def _sleep(seconds: int) -> None:
+        sleep_values.append(seconds)
+        if len(sleep_values) == 3:
+            raise asyncio.CancelledError
+
+    monkeypatch.setattr(rule_sync, "sync_once", _sync_once)
+    monkeypatch.setattr(rule_sync.asyncio, "sleep", _sleep)
+    monkeypatch.setattr(rule_sync, "_FIRST_DELAY_SEC", 1)
+    monkeypatch.setattr(rule_sync, "_LOOP_BACKOFF_SEC", 2)
+    monkeypatch.setattr(rule_sync, "_SYNC_INTERVAL_SEC", 3)
+
+    try:
+        asyncio.run(rule_sync.run_rule_catalog_sync_loop())
+    except asyncio.CancelledError:
+        pass
+    else:
+        raise AssertionError("loop should stop at the test cancellation point")
+
+    assert sync_count == 2
+    assert sleep_values == [1, 2, 3]

apps/api/tests/test_signal_worker_ansible_executor_binding.py

@@ -93,6 +93,7 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None:
     assert worker_env["AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS"] == "168"
     assert worker_env["DATABASE_NULL_POOL"] == "true"
     assert worker_env["DATABASE_NULL_POOL_CONCURRENCY_LIMIT"] == "1"
+    assert worker_env["DATABASE_POOL_TIMEOUT_SECONDS"] == "30"
 
 
 def test_execution_broker_is_only_workload_with_ssh_transport() -> None: