diff --git a/apps/api/src/jobs/asset_capability_reconciliation_job.py b/apps/api/src/jobs/asset_capability_reconciliation_job.py index c4dd07b0b..fbde39a20 100644 --- a/apps/api/src/jobs/asset_capability_reconciliation_job.py +++ b/apps/api/src/jobs/asset_capability_reconciliation_job.py @@ -359,7 +359,7 @@ async def reconcile_once( ) public_cache_refreshed = bool( public_matrix.get("live_readback_status") == "ready" - and public_matrix.get("closed") is True + and public_matrix.get("reconciliation_closed") is True ) receipt["public_cache_refreshed"] = public_cache_refreshed if not public_cache_refreshed: @@ -367,6 +367,7 @@ async def reconcile_once( logger.warning( "asset_capability_reconciliation_public_cache_refresh_degraded", live_readback_status=public_matrix.get("live_readback_status"), + reconciliation_closed=public_matrix.get("reconciliation_closed"), closed=public_matrix.get("closed"), ) receipt["duration_ms"] = int((time.monotonic() - started) * 1000) 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 81233f1af..4c0850273 100644 --- a/apps/api/src/services/ai_automation_asset_capability_matrix.py +++ b/apps/api/src/services/ai_automation_asset_capability_matrix.py @@ -741,6 +741,59 @@ def _matrix_fingerprint_row(asset: Mapping[str, Any]) -> dict[str, Any]: } +def _apply_capability_closure_truth(payload: dict[str, Any]) -> dict[str, Any]: + """Separate reconciliation success from actual capability closure.""" + + controls = _dict(payload.get("controls")) + summary = _dict(payload.get("summary")) + latest_run = _dict(payload.get("latest_discovery_run")) + gap_asset_count = int(summary.get("gap_asset_count") or 0) + covered_stage_cell_count = int(summary.get("covered_stage_cell_count") or 0) + missing_stage_cell_count = int(summary.get("missing_stage_cell_count") or 0) + stale_stage_cell_count = int(summary.get("stale_stage_cell_count") or 0) + stage_cell_count = int(summary.get("stage_cell_count") or 0) + reconciliation_closed = all( + controls.get(control_id) is True for control_id in REQUIRED_CONTROL_IDS + ) + capability_coverage_complete = bool( + int(summary.get("matrix_asset_count") or 0) > 0 + and stage_cell_count > 0 + and gap_asset_count == 0 + and missing_stage_cell_count == 0 + and stale_stage_cell_count == 0 + and covered_stage_cell_count == stage_cell_count + ) + closure_controls = { + "reconciliation_closed": reconciliation_closed, + "live_readback_ready": payload.get("live_readback_status") == "ready", + "latest_discovery_run_fresh": latest_run.get("fresh") is True, + "zero_gap_assets": gap_asset_count == 0, + "zero_missing_stage_cells": missing_stage_cell_count == 0, + "zero_stale_stage_cells": stale_stage_cell_count == 0, + "full_stage_coverage": bool( + stage_cell_count > 0 and covered_stage_cell_count == stage_cell_count + ), + } + capability_closed = capability_coverage_complete and all( + closure_controls.values() + ) + if capability_closed: + status = "closed" + elif not capability_coverage_complete: + status = "capability_gaps_open" + elif closure_controls["live_readback_ready"] is not True: + status = "live_readback_degraded" + else: + status = "reconciliation_required" + + payload["reconciliation_closed"] = reconciliation_closed + payload["capability_coverage_complete"] = capability_coverage_complete + payload["closure_controls"] = closure_controls + payload["closed"] = capability_closed + payload["status"] = status + return payload + + def build_asset_capability_matrix( scope_classes: Sequence[Mapping[str, Any]], *, @@ -762,10 +815,13 @@ def build_asset_capability_matrix( for scope in scope_classes } latest = _dict(latest_run) - run_fresh = _is_fresh( - latest.get("ended_at"), - now=now, - slo_seconds=FRESHNESS_SLO_SECONDS, + run_fresh = bool( + str(latest.get("status") or "") == "success" + and _is_fresh( + latest.get("ended_at"), + now=now, + slo_seconds=FRESHNESS_SLO_SECONDS, + ) ) slug_index, token_index = _build_live_identity_indexes(live_rows) used_live_indexes: set[int] = set() @@ -928,16 +984,12 @@ def build_asset_capability_matrix( == "live_observed" ) total_stage_cells = len(assets) * len(STAGE_IDS) - matrix_closed = all( - controls.get(control_id) is True for control_id in REQUIRED_CONTROL_IDS - ) receipt = _dict(reconciliation_receipt) receipt_output = _dict(receipt.get("output")) - return { + payload = { "schema_version": SCHEMA_VERSION, "generated_at": now.isoformat().replace("+00:00", "Z"), - "status": "closed" if matrix_closed else "reconciliation_required", "work_item_id": "AIA-P0-006", "live_readback_status": live_readback_status, "matrix_fingerprint": matrix_fingerprint, @@ -951,7 +1003,6 @@ def build_asset_capability_matrix( }, "controls": controls, "required_control_ids": list(REQUIRED_CONTROL_IDS), - "closed": matrix_closed, "summary": { "scope_class_count": len(expected_scope_counts), "declared_asset_count": len(declared_assets), @@ -1043,6 +1094,7 @@ def build_asset_capability_matrix( "host_write_performed": False, }, } + return _apply_capability_closure_truth(payload) def build_reconciliation_candidates(matrix: Mapping[str, Any]) -> list[dict[str, Any]]: @@ -1393,7 +1445,7 @@ def _annotate_cached_matrix( if error_type: payload["live_readback_status"] = "degraded_cached" payload["live_readback_error_type"] = error_type - return payload + return _apply_capability_closure_truth(payload) async def build_asset_capability_matrix_with_live_readback( 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 4642c75e2..054953162 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/apps/api/tests/test_ai_automation_asset_capability_matrix.py @@ -419,9 +419,71 @@ def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> N assert set(asset["stages"]) == set(STAGE_IDS) assert {stage["status"] for stage in asset["stages"].values()} == {"covered"} assert all(matrix["controls"][control_id] for control_id in REQUIRED_CONTROL_IDS) + assert matrix["reconciliation_closed"] is True + assert matrix["capability_coverage_complete"] is True + assert all(matrix["closure_controls"].values()) assert matrix["closed"] is True + assert matrix["status"] == "closed" assert matrix["summary"]["gap_asset_count"] == 0 + degraded = 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=receipt, + repo_root=_REPO_ROOT, + generated_at=now, + live_readback_status="degraded", + ) + assert degraded["reconciliation_closed"] is True + assert degraded["capability_coverage_complete"] is True + assert degraded["closure_controls"]["live_readback_ready"] is False + assert degraded["closed"] is False + assert degraded["status"] == "live_readback_degraded" + + +def test_closed_reconciliation_receipt_cannot_hide_capability_gaps() -> None: + now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC) + live = _fully_covered_host(now) + live["coverage"]["auto_remediation"]["coverage_status"] = "red" + initial = build_asset_capability_matrix( + _single_host_scope(), + live_assets=[live], + 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=[live], + 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": 1, + "repository_readback_count": 1, + "work_item_verifier_readback_count": 1, + "repository_readback_verified": True, + "closed": True, + }, + }, + repo_root=_REPO_ROOT, + generated_at=now, + ) + + assert all(matrix["controls"][control_id] for control_id in REQUIRED_CONTROL_IDS) + assert matrix["reconciliation_closed"] is True + assert matrix["capability_coverage_complete"] is False + assert matrix["closure_controls"]["zero_gap_assets"] is False + assert matrix["closure_controls"]["zero_missing_stage_cells"] is False + assert matrix["summary"]["gap_asset_count"] == 1 + assert matrix["summary"]["missing_stage_cell_count"] > 0 + assert matrix["closed"] is False + assert matrix["status"] == "capability_gaps_open" + def test_reconciliation_receipt_requires_closed_terminal() -> None: now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC) @@ -667,7 +729,12 @@ def test_reconcile_once_publishes_closed_public_matrix_cache( "ended_at": settled_at, }, } - return {"live_readback_status": "ready", "closed": True} + return { + "live_readback_status": "ready", + "reconciliation_closed": True, + "closed": False, + "status": "capability_gaps_open", + } async def _no_wait(_seconds: int) -> None: return None @@ -1292,6 +1359,7 @@ def test_public_matrix_timeout_uses_last_ready_cache(monkeypatch: Any) -> None: ) assert payload["live_readback_status"] == "degraded_cached" + assert payload["closed"] is False assert payload["live_readback_error_type"] == "TimeoutError" assert payload["summary"]["live_inventory_asset_count"] == 9595 assert payload["summary"]["live_observed_asset_count"] == 193