From e01db7391e89a52958d9c3f06c3218fdc5053eb7 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 09:11:50 +0800 Subject: [PATCH] fix(automation): close asset reconciliation controls --- .../asset_capability_reconciliation_job.py | 290 +++++++++++------- .../ai_automation_asset_capability_matrix.py | 12 +- ...t_ai_automation_asset_capability_matrix.py | 176 ++++++++++- 3 files changed, 357 insertions(+), 121 deletions(-) diff --git a/apps/api/src/jobs/asset_capability_reconciliation_job.py b/apps/api/src/jobs/asset_capability_reconciliation_job.py index c13545927..446da976e 100644 --- a/apps/api/src/jobs/asset_capability_reconciliation_job.py +++ b/apps/api/src/jobs/asset_capability_reconciliation_job.py @@ -1,4 +1,4 @@ -"""Daily AIA-P0-006 asset capability reconciliation. +"""AIA-P0-006 asset capability reconciliation. The job projects added, removed, drifted, missing, and stale asset capability states into idempotent automation_operation_log work-item receipts. It writes no @@ -13,9 +13,7 @@ import time import uuid from collections import Counter from collections.abc import Mapping, Sequence -from datetime import datetime, timedelta from typing import Any -from zoneinfo import ZoneInfo import structlog @@ -31,18 +29,17 @@ from src.services.ai_automation_asset_capability_matrix import ( logger = structlog.get_logger(__name__) -_FIRST_DELAY_SECONDS = 90 -_LOOP_BACKOFF_SECONDS = 30 * 60 -_MAX_RECONCILIATION_CANDIDATES = 20_000 +_FIRST_DELAY_SECONDS = 30 +_LOOP_BACKOFF_SECONDS = 5 * 60 +_RECONCILIATION_POLL_SECONDS = 5 * 60 _RECONCILIATION_LIVE_READBACK_TIMEOUT_SECONDS = 30.0 +_WORK_ITEM_INSERT_BATCH_SIZE = 250 +_MAX_RECONCILIATION_CANDIDATES = 20_000 _RETRYABLE_RESULT_STATUSES = { "blocked_safe_no_write", "degraded_cache_refresh", "degraded_no_write", } -_DAILY_TRIGGER_HOUR_TAIPEI = 3 -_DAILY_TRIGGER_MINUTE_TAIPEI = 30 -_TAIPEI = ZoneInfo("Asia/Taipei") def _dict(value: Any) -> dict[str, Any]: @@ -53,19 +50,6 @@ def _canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) -def _seconds_until_next_trigger() -> int: - now = datetime.now(_TAIPEI) - target = now.replace( - hour=_DAILY_TRIGGER_HOUR_TAIPEI, - minute=_DAILY_TRIGGER_MINUTE_TAIPEI, - second=0, - microsecond=0, - ) - if target <= now: - target += timedelta(days=1) - return max(60, int((target - now).total_seconds())) - - def build_reconciliation_summary_payload( matrix: Mapping[str, Any], candidates: Sequence[Mapping[str, Any]], @@ -74,6 +58,7 @@ def build_reconciliation_summary_payload( automation_run_id: str, work_item_id: str, repository_readback_count: int, + work_item_verifier_readback_count: int, created_count: int, existing_count: int, ) -> dict[str, Any]: @@ -89,7 +74,10 @@ def build_reconciliation_summary_payload( "full_scope_class_coverage", ) candidate_count = len(candidates) - readback_verified = repository_readback_count == candidate_count + readback_verified = bool( + repository_readback_count == candidate_count + and work_item_verifier_readback_count == candidate_count + ) mechanism_closed = bool( all(controls.get(control_id) is True for control_id in structural_control_ids) and str(_dict(matrix.get("latest_discovery_run")).get("run_id") or "") @@ -110,6 +98,7 @@ def build_reconciliation_summary_payload( "created_work_item_count": created_count, "existing_work_item_count": existing_count, "repository_readback_count": repository_readback_count, + "work_item_verifier_readback_count": work_item_verifier_readback_count, "repository_readback_verified": readback_verified, "structural_controls": { control_id: controls.get(control_id) is True @@ -135,6 +124,7 @@ def build_reconciliation_summary_payload( "bounded_execution_receipt": { "target": "automation_operation_log", "operation": "idempotent_work_item_projection", + "batch_size": _WORK_ITEM_INSERT_BATCH_SIZE, "created_count": created_count, "existing_count": existing_count, }, @@ -142,6 +132,7 @@ def build_reconciliation_summary_payload( "verifier": "automation_operation_log_row_readback", "expected_count": candidate_count, "observed_count": repository_readback_count, + "verified_work_item_count": work_item_verifier_readback_count, "passed": readback_verified, }, "rollback_or_no_write_terminal": "no_external_runtime_write", @@ -186,12 +177,11 @@ def build_reconciliation_work_item_input( async def run_asset_capability_reconciliation_loop() -> None: - """Run shortly after startup, then at 03:30 Asia/Taipei every day.""" + """Reconcile shortly after startup and retry each discovery fingerprint.""" logger.info( "asset_capability_reconciliation_loop_started", - daily_trigger_hour_taipei=_DAILY_TRIGGER_HOUR_TAIPEI, - daily_trigger_minute_taipei=_DAILY_TRIGGER_MINUTE_TAIPEI, + poll_interval_seconds=_RECONCILIATION_POLL_SECONDS, ) await asyncio.sleep(_FIRST_DELAY_SECONDS) triggered_by = "startup" @@ -216,8 +206,8 @@ async def run_asset_capability_reconciliation_loop() -> None: await asyncio.sleep(_LOOP_BACKOFF_SECONDS) triggered_by = "retry" continue - await asyncio.sleep(_seconds_until_next_trigger()) - triggered_by = "cron" + await asyncio.sleep(_RECONCILIATION_POLL_SECONDS) + triggered_by = "poll" async def reconcile_once( @@ -227,13 +217,6 @@ async def reconcile_once( ) -> dict[str, Any]: """Reconcile the matrix and persist idempotent gap/drift work items.""" - from src.services.ai_advisory_helpers import try_acquire_daily_lock - - if triggered_by == "cron" and not await try_acquire_daily_lock( - "asset_capability_reconciliation" - ): - return {"status": "skipped", "reason": "not_leader"} - started = time.monotonic() matrix = await build_asset_capability_matrix_with_live_readback( project_id=project_id, @@ -247,6 +230,7 @@ async def reconcile_once( ) return { "status": "degraded_no_write", + "triggered_by": triggered_by, "live_readback_status": matrix.get("live_readback_status"), "candidate_count": 0, } @@ -310,17 +294,43 @@ async def _persist_reconciliation( from src.db.base import get_db_context matrix_fingerprint = str(matrix.get("matrix_fingerprint") or "") - taipei_date = datetime.now(_TAIPEI).date().isoformat() + discovery_run_id = str( + _dict(matrix.get("latest_discovery_run")).get("run_id") or "" + ) + candidate_rows = [dict(candidate) for candidate in candidates] + expected_fingerprint_list = [ + str(candidate.get("candidate_fingerprint") or "") + for candidate in candidate_rows + ] + expected_fingerprints = set(expected_fingerprint_list) + if not matrix_fingerprint: + raise ValueError("reconciliation_matrix_fingerprint_missing") + if not discovery_run_id: + raise ValueError("reconciliation_discovery_run_missing") + if ( + any(not fingerprint for fingerprint in expected_fingerprint_list) + or len(expected_fingerprints) != len(candidate_rows) + ): + raise ValueError("reconciliation_candidate_fingerprint_invalid") + idempotency_key = ( - f"asset-capability-reconciliation:{taipei_date}:{matrix_fingerprint}" + "asset-capability-reconciliation:" + f"{discovery_run_id}:{matrix_fingerprint}" ) trace_id = f"asset-capability:{matrix_fingerprint[:16]}" automation_run_id = str(uuid.uuid4()) work_item_id = "AIA-P0-006" - discovery_run_id = str( - _dict(matrix.get("latest_discovery_run")).get("run_id") or "" + fingerprint_readback_sql = text( + """ + SELECT DISTINCT input ->> 'candidate_fingerprint' AS candidate_fingerprint + FROM automation_operation_log + WHERE actor = :actor + AND COALESCE(input ->> 'semantic_operation_type', operation_type) + = :semantic_operation_type + AND input ->> 'candidate_fingerprint' + = ANY(CAST(:candidate_fingerprints AS text[])) + """ ) - created_count = 0 async with get_db_context(project_id) as db: await db.execute( @@ -338,6 +348,9 @@ async def _persist_reconciliation( AND input ->> 'idempotency_key' = :idempotency_key AND status = 'success' AND output ->> 'closed' = 'true' + AND output ->> 'repository_readback_verified' = 'true' + AND output ->> 'candidate_count' + = output ->> 'work_item_verifier_readback_count' ORDER BY created_at DESC LIMIT 1 """ @@ -352,38 +365,23 @@ async def _persist_reconciliation( if isinstance(prior, dict): return prior - candidate_fingerprints = [ - str(candidate.get("candidate_fingerprint") or "") - for candidate in candidates - ] - existing_fingerprints: set[str] = set() - if candidate_fingerprints: - existing = await db.execute( - text( - """ - SELECT DISTINCT input ->> 'candidate_fingerprint' - FROM automation_operation_log - WHERE actor = :actor - AND COALESCE(input ->> 'semantic_operation_type', operation_type) - = :semantic_operation_type - AND input ->> 'candidate_fingerprint' - = ANY(CAST(:candidate_fingerprints AS text[])) - """ - ), + persisted_before: set[str] = set() + if expected_fingerprint_list: + existing_result = await db.execute( + fingerprint_readback_sql, { "actor": RECONCILIATION_ACTOR, "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, - "candidate_fingerprints": candidate_fingerprints, + "candidate_fingerprints": expected_fingerprint_list, }, ) - existing_fingerprints = { - str(row[0]) for row in existing.fetchall() if row[0] + persisted_before = { + str(row[0]) for row in existing_result.fetchall() if row[0] } - insert_rows: list[dict[str, Any]] = [] - for candidate_raw in candidates: - candidate = dict(candidate_raw) - if candidate["candidate_fingerprint"] in existing_fingerprints: + insert_parameters: list[dict[str, Any]] = [] + for candidate in candidate_rows: + if candidate["candidate_fingerprint"] in persisted_before: continue input_payload = build_reconciliation_work_item_input( candidate, @@ -398,7 +396,7 @@ async def _persist_reconciliation( "post_verifier": "pending_batch_readback", "rollback_or_no_write_terminal": "no_external_runtime_write", } - insert_rows.append( + insert_parameters.append( { "asset_id": candidate.get("asset_inventory_id"), "discovery_run_id": discovery_run_id, @@ -409,7 +407,9 @@ async def _persist_reconciliation( { "would_create_work_item": True, "external_runtime_write": False, - "candidate_fingerprint": candidate["candidate_fingerprint"], + "candidate_fingerprint": candidate[ + "candidate_fingerprint" + ], } ), "tags": [ @@ -420,74 +420,134 @@ async def _persist_reconciliation( } ) - if insert_rows: - await db.execute( - text( - """ - INSERT INTO automation_operation_log ( - operation_type, - asset_id, - run_id, - actor, - input, - output, - dry_run_result, - status, - tags - ) VALUES ( - 'asset_discovered', - :asset_id, - CAST(NULLIF(:discovery_run_id, '') AS uuid), - :actor, - CAST(:input AS jsonb), - CAST(:output AS jsonb), - CAST(:dry_run_result AS jsonb), - 'success', - :tags - ) - """ - ), - insert_rows, + insert_work_item_sql = text( + """ + INSERT INTO automation_operation_log ( + operation_type, + asset_id, + run_id, + actor, + input, + output, + dry_run_result, + status, + tags + ) VALUES ( + 'asset_discovered', + :asset_id, + CAST(NULLIF(:discovery_run_id, '') AS uuid), + :actor, + CAST(:input AS jsonb), + CAST(:output AS jsonb), + CAST(:dry_run_result AS jsonb), + 'success', + :tags + ) + """ + ) + for batch_start in range( + 0, + len(insert_parameters), + _WORK_ITEM_INSERT_BATCH_SIZE, + ): + await db.execute( + insert_work_item_sql, + insert_parameters[ + batch_start : batch_start + _WORK_ITEM_INSERT_BATCH_SIZE + ], ) - created_count = len(insert_rows) - expected_fingerprints = { - str(candidate.get("candidate_fingerprint") or "") - for candidate in candidates - } persisted_fingerprints: set[str] = set() - if candidate_fingerprints: + if expected_fingerprint_list: rows_result = await db.execute( - text( - """ - SELECT DISTINCT input ->> 'candidate_fingerprint' - FROM automation_operation_log - WHERE actor = :actor - AND COALESCE(input ->> 'semantic_operation_type', operation_type) - = :semantic_operation_type - AND input ->> 'candidate_fingerprint' - = ANY(CAST(:candidate_fingerprints AS text[])) - """ - ), + fingerprint_readback_sql, { "actor": RECONCILIATION_ACTOR, "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, - "candidate_fingerprints": candidate_fingerprints, + "candidate_fingerprints": expected_fingerprint_list, }, ) persisted_fingerprints = { str(row[0]) for row in rows_result.fetchall() if row[0] } repository_readback_count = len(expected_fingerprints & persisted_fingerprints) + work_item_verifier_readback_count = 0 + if repository_readback_count == len(expected_fingerprints): + if expected_fingerprint_list: + await db.execute( + text( + """ + UPDATE automation_operation_log + SET output = COALESCE(output, '{}'::jsonb) + || CAST(:verification_output AS jsonb) + WHERE actor = :actor + AND COALESCE( + input ->> 'semantic_operation_type', operation_type + ) = :semantic_operation_type + AND input ->> 'candidate_fingerprint' + = ANY(CAST(:candidate_fingerprints AS text[])) + AND COALESCE( + output ->> 'repository_readback_verified', 'false' + ) <> 'true' + """ + ), + { + "actor": RECONCILIATION_ACTOR, + "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, + "candidate_fingerprints": expected_fingerprint_list, + "verification_output": _canonical_json( + { + "post_verifier": "passed_repository_readback", + "repository_readback_verified": True, + "verified_by": ( + "asset_capability_reconciler_batch_readback" + ), + "verified_run_id": automation_run_id, + } + ), + }, + ) + verified_result = await db.execute( + text( + """ + SELECT DISTINCT + input ->> 'candidate_fingerprint' AS candidate_fingerprint + FROM automation_operation_log + WHERE actor = :actor + AND COALESCE( + input ->> 'semantic_operation_type', operation_type + ) = :semantic_operation_type + AND input ->> 'candidate_fingerprint' + = ANY(CAST(:candidate_fingerprints AS text[])) + AND output ->> 'repository_readback_verified' = 'true' + """ + ), + { + "actor": RECONCILIATION_ACTOR, + "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, + "candidate_fingerprints": expected_fingerprint_list, + }, + ) + verified_fingerprints = { + str(row[0]) for row in verified_result.fetchall() if row[0] + } + work_item_verifier_readback_count = len( + expected_fingerprints & verified_fingerprints + ) + + created_count = len(insert_parameters) summary_output = build_reconciliation_summary_payload( matrix, - candidates, + candidate_rows, trace_id=trace_id, automation_run_id=automation_run_id, work_item_id=work_item_id, repository_readback_count=repository_readback_count, + work_item_verifier_readback_count=( + work_item_verifier_readback_count + ), created_count=created_count, - existing_count=len(candidates) - created_count, + existing_count=len(candidate_rows) - created_count, ) summary_input = { "schema_version": RECONCILIATION_SCHEMA_VERSION, diff --git a/apps/api/src/services/ai_automation_asset_capability_matrix.py b/apps/api/src/services/ai_automation_asset_capability_matrix.py index 30ab3ac40..6bd56a430 100644 --- a/apps/api/src/services/ai_automation_asset_capability_matrix.py +++ b/apps/api/src/services/ai_automation_asset_capability_matrix.py @@ -627,6 +627,9 @@ def _receipt_controls( output = _dict(row.get("output")) candidate_count = int(output.get("candidate_count") or 0) readback_count = int(output.get("repository_readback_count") or 0) + verifier_readback_count = int( + output.get("work_item_verifier_readback_count") or 0 + ) created_at = row.get("created_at") fresh = bool( row @@ -647,6 +650,7 @@ def _receipt_controls( fresh and output.get("repository_readback_verified") is True and candidate_count == readback_count + and candidate_count == verifier_readback_count ), } @@ -912,6 +916,9 @@ def build_asset_capability_matrix( "reconciliation_work_item_readback_count": int( receipt_output.get("repository_readback_count") or 0 ), + "reconciliation_work_item_verifier_readback_count": int( + receipt_output.get("work_item_verifier_readback_count") or 0 + ), }, "scope_rollups": [ { @@ -945,6 +952,9 @@ def build_asset_capability_matrix( "repository_readback_verified" ) is True, + "work_item_verifier_readback_count": int( + receipt_output.get("work_item_verifier_readback_count") or 0 + ), }, "operation_boundaries": { "read_only_api": True, @@ -1074,7 +1084,7 @@ async def _load_live_rows( cs.coverage_status, cs.created_at AS coverage_created_at FROM asset_inventory ai - LEFT JOIN asset_coverage_snapshot cs + JOIN asset_coverage_snapshot cs ON cs.asset_id = ai.asset_id AND cs.run_id = CAST(NULLIF(:run_id, '') AS uuid) WHERE ai.lifecycle_state <> 'decommissioned' diff --git a/apps/api/tests/test_ai_automation_asset_capability_matrix.py b/apps/api/tests/test_ai_automation_asset_capability_matrix.py index ac817d4c5..0e9d3185f 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/apps/api/tests/test_ai_automation_asset_capability_matrix.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import inspect +import json from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any @@ -214,6 +215,7 @@ def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> N "matrix_fingerprint": initial["matrix_fingerprint"], "candidate_count": 0, "repository_readback_count": 0, + "work_item_verifier_readback_count": 0, "repository_readback_verified": True, "closed": True, }, @@ -334,6 +336,7 @@ def test_reconciliation_summary_requires_repository_readback() -> None: automation_run_id="run-1", work_item_id="AIA-P0-006", repository_readback_count=0, + work_item_verifier_readback_count=0, created_count=1, existing_count=0, ) @@ -344,6 +347,7 @@ def test_reconciliation_summary_requires_repository_readback() -> None: automation_run_id="run-1", work_item_id="AIA-P0-006", repository_readback_count=1, + work_item_verifier_readback_count=1, created_count=1, existing_count=0, ) @@ -354,6 +358,31 @@ def test_reconciliation_summary_requires_repository_readback() -> None: assert passed["closed"] is True +def test_reconciliation_summary_requires_each_work_item_verifier() -> None: + matrix = build_asset_capability_matrix( + _single_host_scope(), + latest_run={"run_id": "discovery-run-1", "status": "success"}, + repo_root=_REPO_ROOT, + ) + candidates = build_reconciliation_candidates(matrix) + + payload = build_reconciliation_summary_payload( + matrix, + candidates, + trace_id="trace-1", + automation_run_id="run-1", + work_item_id="AIA-P0-006", + repository_readback_count=1, + work_item_verifier_readback_count=0, + created_count=1, + existing_count=0, + ) + + assert payload["repository_readback_verified"] is False + assert payload["post_verifier"]["passed"] is False + assert payload["closed"] is False + + def test_reconciliation_work_item_uses_same_trace_and_run_as_summary() -> None: matrix = build_asset_capability_matrix( _single_host_scope(), @@ -380,13 +409,15 @@ 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:") + loop_start = source.index("for candidate in candidate_rows:") + loop_end = source.index("insert_work_item_sql =") assert "ANY(CAST(:candidate_fingerprints AS text[]))" in source - assert "insert_rows" in source + assert "insert_parameters" in source + assert "_WORK_ITEM_INSERT_BATCH_SIZE" in source assert "await db.execute" not in source[loop_start:loop_end] assert "AND output ->> 'closed' = 'true'" in source + assert "work_item_verifier_readback_count" in source def test_reconcile_once_uses_bounded_worker_live_readback_timeout( @@ -474,7 +505,7 @@ def test_reconcile_once_publishes_closed_public_matrix_cache( assert result["public_cache_refreshed"] is True -def test_reconciliation_loop_retries_degraded_result_before_daily_wait( +def test_reconciliation_loop_retries_degraded_result_before_poll( monkeypatch: Any, ) -> None: triggered_by_values: list[str] = [] @@ -495,7 +526,7 @@ def test_reconciliation_loop_retries_degraded_result_before_daily_wait( 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) + monkeypatch.setattr(reconciliation_job, "_RECONCILIATION_POLL_SECONDS", 1) try: asyncio.run(reconciliation_job.run_asset_capability_reconciliation_loop()) @@ -624,6 +655,136 @@ class _FakeDbContext: return None +class _FakeReconciliationResult: + def __init__( + self, + *, + scalar: Any = None, + rows: list[tuple[Any, ...]] | None = None, + ) -> None: + self._scalar = scalar + self._rows = rows or [] + + def scalar_one_or_none(self) -> Any: + return self._scalar + + def scalar_one(self) -> Any: + return self._scalar + + def fetchall(self) -> list[tuple[Any, ...]]: + return self._rows + + +class _FakeReconciliationDb: + def __init__(self) -> None: + self.persisted_fingerprints: set[str] = set() + self.verified_fingerprints: set[str] = set() + self.summary_outputs: list[dict[str, Any]] = [] + self.work_item_batch_sizes: list[int] = [] + + async def execute( + self, + statement: Any, + params: dict[str, Any] | list[dict[str, Any]] | None = None, + ) -> _FakeReconciliationResult: + sql = str(statement) + if "pg_advisory_xact_lock" in sql: + return _FakeReconciliationResult() + if "SELECT output" in sql: + prior = self.summary_outputs[-1] if self.summary_outputs else None + return _FakeReconciliationResult(scalar=prior) + if "INSERT INTO automation_operation_log" in sql and isinstance(params, list): + self.work_item_batch_sizes.append(len(params)) + for row in params: + payload = json.loads(str(row["input"])) + self.persisted_fingerprints.add(payload["candidate_fingerprint"]) + return _FakeReconciliationResult() + if "UPDATE automation_operation_log" in sql: + assert isinstance(params, dict) + requested = set(params["candidate_fingerprints"]) + self.verified_fingerprints.update( + requested & self.persisted_fingerprints + ) + return _FakeReconciliationResult() + if "SELECT DISTINCT" in sql and "candidate_fingerprint" in sql: + assert isinstance(params, dict) + requested = set(params["candidate_fingerprints"]) + source = ( + self.verified_fingerprints + if "repository_readback_verified" in sql + else self.persisted_fingerprints + ) + return _FakeReconciliationResult( + rows=[(fingerprint,) for fingerprint in sorted(requested & source)] + ) + if "RETURNING op_id::text" in sql: + assert isinstance(params, dict) + self.summary_outputs.append( + json.loads(str(params["output"])) + ) + return _FakeReconciliationResult(scalar="summary-op-1") + raise AssertionError(f"unexpected SQL: {sql}") + + +class _FakeReconciliationDbContext: + def __init__(self, db: _FakeReconciliationDb) -> None: + self.db = db + + async def __aenter__(self) -> _FakeReconciliationDb: + return self.db + + async def __aexit__(self, *_args: object) -> None: + return None + + +def test_reconciliation_persists_bounded_batches_and_reuses_verified_summary( + monkeypatch: Any, +) -> None: + db = _FakeReconciliationDb() + monkeypatch.setattr( + "src.db.base.get_db_context", + lambda _project_id: _FakeReconciliationDbContext(db), + ) + matrix = { + "matrix_fingerprint": "matrix-fingerprint-1", + "latest_discovery_run": { + "run_id": "00000000-0000-0000-0000-000000000001" + }, + "controls": { + control_id: True + for control_id in REQUIRED_CONTROL_IDS[:6] + }, + } + candidates = [ + { + "candidate_fingerprint": f"fingerprint-{index:04d}", + "candidate_kind": "capability_gap", + "canonical_asset_id": f"awoooi:services:asset-{index:04d}", + "asset_inventory_id": index, + "missing_stage_ids": ["sensor"], + "stale_stage_ids": [], + "change_types": [], + } + for index in range(301) + ] + + first = asyncio.run( + _persist_reconciliation(matrix, candidates, project_id="awoooi") + ) + second = asyncio.run( + _persist_reconciliation(matrix, candidates, project_id="awoooi") + ) + + assert db.work_item_batch_sizes == [250, 51] + assert first["candidate_count"] == 301 + assert first["repository_readback_count"] == 301 + assert first["work_item_verifier_readback_count"] == 301 + assert first["repository_readback_verified"] is True + assert first["closed"] is True + assert second["repository_readback_verified"] is True + assert len(db.summary_outputs) == 1 + + def test_live_change_event_readback_is_scoped_to_selected_asset_ids( monkeypatch: Any, ) -> None: @@ -647,6 +808,11 @@ def test_live_change_event_readback_is_scoped_to_selected_asset_ids( change_sql, change_params = next( (sql, params) for sql, params in db.calls if "FROM asset_change_event" in sql ) + inventory_sql = next( + sql for sql, _params in db.calls if "FROM asset_inventory" in sql + ) + assert "JOIN asset_coverage_snapshot" in inventory_sql + assert "LEFT JOIN asset_coverage_snapshot" not in inventory_sql assert "asset_id = ANY(:asset_ids)" in change_sql assert change_params == {"asset_ids": [42]} assert assets[0]["change_types"] == ["asset_modified"]