fix(assets): separate reconciliation from capability closure

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)

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(