awoooi/apps/api/src/services/ai_automation_asset_capability_matrix.py

"""Canonical all-asset capability matrix for AIA-P0-006.

This module reconciles the committed program scope with ADR-090 live asset,
coverage, compliance, and change receipts. Public payloads only expose stable
canonical identities and fingerprints; raw asset keys and host values stay in
the database boundary.
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import re
from collections import Counter, defaultdict
from collections.abc import Iterable, Mapping, Sequence
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

import structlog

logger = structlog.get_logger(__name__)

SCHEMA_VERSION = "ai_automation_asset_capability_matrix_v1"
RECONCILIATION_SCHEMA_VERSION = "asset_capability_reconciliation_v1"
DEFAULT_PROJECT_ID = "awoooi"
RECONCILIATION_ACTOR = "asset_capability_reconciler"
RECONCILIATION_SUMMARY_OPERATION = "asset_capability_reconciliation_completed"
RECONCILIATION_WORK_ITEM_OPERATION = "asset_capability_reconciliation_work_item_created"
FRESHNESS_SLO_SECONDS = 2 * 60 * 60
RECONCILIATION_FRESHNESS_SLO_SECONDS = 36 * 60 * 60
LIVE_READBACK_TIMEOUT_SECONDS = 8.0
LIVE_READBACK_DB_STATEMENT_TIMEOUT_MILLISECONDS = 6_000

STAGE_IDS = (
    "source_truth",
    "runtime_identity",
    "sensor",
    "policy",
    "executor",
    "verifier",
    "backup_restore",
    "learning_writeback",
)

REQUIRED_CONTROL_IDS = (
    "canonical_id_per_asset",
    "owner_lane_per_asset",
    "source_truth_per_asset",
    "runtime_identity_per_asset",
    "per_asset_stage_status",
    "full_scope_class_coverage",
    "daily_reconciliation_receipt",
    "reconciliation_matrix_fingerprint_match",
    "drift_work_items_verified",
)

_STRUCTURAL_CONTROL_IDS = REQUIRED_CONTROL_IDS[:6]
_OWNER_LANES = {
    "hosts": "platform_sre",
    "services": "runtime_platform",
    "products": "product_governance",
    "websites": "web_reliability",
    "tools": "ai_platform",
    "packages": "software_supply_chain",
    "data_and_backup": "data_resilience",
    "observability_and_security": "aisoc",
}
_LIVE_TYPE_SCOPE = {
    "host": "hosts",
    "container": "services",
    "k8s_workload": "services",
    "k8s_resource": "services",
    "database": "data_and_backup",
    "table": "data_and_backup",
    "website": "websites",
    "api_endpoint": "websites",
    "package": "packages",
    "log_stream": "observability_and_security",
    "km_entry": "tools",
    "frontend": "services",
    "backend": "services",
    "ci_pipeline": "tools",
    "gitea_repo": "products",
    "monitoring_target": "observability_and_security",
    "secret": "services",
    "volume": "data_and_backup",
    "network": "services",
    "certificate": "websites",
    "scheduled_job": "services",
    "message_queue": "services",
    "cache": "data_and_backup",
    "dashboard": "observability_and_security",
    "ai_agent": "tools",
    "llm_model": "tools",
    "third_party_service": "tools",
    "backup_target": "data_and_backup",
}
_STAGE_DIMENSIONS = {
    "sensor": ("auto_monitoring", "auto_alerting"),
    "policy": ("auto_rule_creation", "auto_rule_matching"),
    "executor": ("auto_playbook", "auto_remediation"),
    "verifier": ("auto_remediation", "auto_monitoring"),
    "learning_writeback": ("auto_km_creation", "auto_playbook"),
}


def _repo_root() -> Path:
    source_path = Path(__file__).resolve()
    for parent in source_path.parents:
        if (parent / "docs").is_dir() and (parent / "apps").is_dir():
            return parent
    return source_path.parents[2]


def _dict(value: Any) -> dict[str, Any]:
    return value if isinstance(value, dict) else {}


def _list(value: Any) -> list[Any]:
    return value if isinstance(value, list) else []


def _strings(value: Any) -> list[str]:
    return [
        item.strip() for item in _list(value) if isinstance(item, str) and item.strip()
    ]


def _canonical_json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"))


def _fingerprint(value: Any, *, length: int = 64) -> str:
    return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest()[:length]


def _slug(value: Any) -> str:
    normalized = re.sub(r"[^a-z0-9]+", "-", str(value or "").lower()).strip("-")
    return normalized or "unnamed"


def _identity_tokens(value: Any) -> set[str]:
    return {
        token for token in re.split(r"[^a-z0-9]+", str(value or "").lower()) if token
    }


def _parse_datetime(value: Any) -> datetime | None:
    if isinstance(value, datetime):
        parsed = value
    elif isinstance(value, str) and value.strip():
        try:
            parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
        except ValueError:
            return None
    else:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=UTC)
    return parsed.astimezone(UTC)


def _iso(value: Any) -> str:
    parsed = _parse_datetime(value)
    return parsed.isoformat().replace("+00:00", "Z") if parsed else ""


def _age_seconds(value: Any, *, now: datetime) -> int | None:
    parsed = _parse_datetime(value)
    if parsed is None:
        return None
    return max(0, int((now - parsed).total_seconds()))


def _read_json(path: Path) -> dict[str, Any]:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    return payload if isinstance(payload, dict) else {}


def _manifest_dependencies(path: Path) -> list[tuple[str, str]]:
    payload = _read_json(path)
    dependencies: list[tuple[str, str]] = []
    for group in (
        "dependencies",
        "devDependencies",
        "peerDependencies",
        "optionalDependencies",
    ):
        values = _dict(payload.get(group))
        dependencies.extend((name, group) for name in sorted(values) if name)
    return dependencies


def _declared_entry(
    *,
    scope: Mapping[str, Any],
    declared_asset_id: str,
    display_name: str | None = None,
    source_refs: Iterable[str] = (),
    identity_suffix: str | None = None,
) -> dict[str, Any]:
    scope_id = str(scope.get("id") or "unknown")
    identity = identity_suffix or declared_asset_id
    refs = list(
        dict.fromkeys(
            [
                *_strings(scope.get("inventory_sources")),
                *(ref for ref in source_refs if ref),
            ]
        )
    )
    return {
        "canonical_id": f"awoooi:{scope_id}:{_slug(identity)}",
        "scope_id": scope_id,
        "scope_label": str(scope.get("label") or scope_id),
        "declared_asset_id": declared_asset_id,
        "display_name": display_name or declared_asset_id,
        "owner_lane": _OWNER_LANES.get(scope_id, "platform_governance"),
        "source_truth_refs": refs,
        "declared": True,
    }


def _package_declared_assets(
    scope: Mapping[str, Any],
    *,
    root: Path,
) -> list[dict[str, Any]]:
    supply_path = (
        root / "docs/evaluations/package_supply_chain_inventory_2026-06-04.json"
    )
    javascript_path = (
        root / "docs/evaluations/javascript_package_inventory_2026-06-04.json"
    )
    supply = _read_json(supply_path)
    javascript = _read_json(javascript_path)
    assets: list[dict[str, Any]] = []

    for surface in _list(supply.get("surfaces")):
        row = _dict(surface)
        surface_id = str(row.get("surface_id") or "")
        if not surface_id:
            continue
        assets.append(
            _declared_entry(
                scope=scope,
                declared_asset_id=surface_id,
                display_name=str(row.get("display_name") or surface_id),
                source_refs=[str(row.get("manifest_ref") or "")],
            )
        )

    dependency_units: list[dict[str, Any]] = []
    for workspace in _list(javascript.get("workspaces")):
        row = _dict(workspace)
        workspace_id = str(row.get("workspace_id") or "workspace")
        manifest_ref = str(row.get("manifest_ref") or "")
        dependencies = (
            _manifest_dependencies(root / manifest_ref) if manifest_ref else []
        )
        if not dependencies:
            expected = int(_dict(row.get("dependency_counts")).get("total") or 0)
            dependencies = [
                (f"declared-dependency-{index:02d}", "inventory_count")
                for index in range(1, expected + 1)
            ]
        for dependency_name, dependency_group in dependencies:
            unit_id = f"{workspace_id}:{dependency_group}:{dependency_name}"
            dependency_units.append(
                _declared_entry(
                    scope=scope,
                    declared_asset_id=unit_id,
                    display_name=f"{workspace_id} / {dependency_name}",
                    source_refs=[manifest_ref],
                    identity_suffix=f"{workspace_id}-dependency-{dependency_name}",
                )
            )

    assets.extend(dependency_units)
    return _fit_declared_count(scope, assets)


def _data_backup_declared_assets(
    scope: Mapping[str, Any],
    *,
    root: Path,
) -> list[dict[str, Any]]:
    assets: list[dict[str, Any]] = []
    product_path = (
        root / "docs/operations/gitea-all-product-dev-prod-repo-readback.snapshot.json"
    )
    products = _list(_read_json(product_path).get("products"))
    for item in _strings(scope.get("asset_ids")):
        if item == "gitea_repo_bundles:12":
            for product in products:
                row = _dict(product)
                product_id = str(row.get("product_id") or "")
                if product_id:
                    assets.append(
                        _declared_entry(
                            scope=scope,
                            declared_asset_id=f"gitea_repo_bundle:{product_id}",
                            display_name=f"Gitea bundle / {product_id}",
                            source_refs=[str(product_path.relative_to(root))],
                        )
                    )
            continue
        assets.append(_declared_entry(scope=scope, declared_asset_id=item))
    return _fit_declared_count(scope, assets)


def _fit_declared_count(
    scope: Mapping[str, Any],
    assets: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]:
    expected = int(scope.get("asset_count") or len(assets))
    fitted = list(assets[:expected])
    for index in range(len(fitted) + 1, expected + 1):
        fitted.append(
            _declared_entry(
                scope=scope,
                declared_asset_id=f"declared-unit-{index:03d}",
                display_name=f"Declared unit {index:03d}",
            )
        )
    return fitted


def build_declared_asset_catalog(
    scope_classes: Sequence[Mapping[str, Any]],
    *,
    repo_root: Path | None = None,
) -> list[dict[str, Any]]:
    """Expand aggregate declarations into one canonical row per asset unit."""

    root = repo_root or _repo_root()
    catalog: list[dict[str, Any]] = []
    for scope in scope_classes:
        scope_id = str(scope.get("id") or "")
        if scope_id == "packages":
            catalog.extend(_package_declared_assets(scope, root=root))
        elif scope_id == "data_and_backup":
            catalog.extend(_data_backup_declared_assets(scope, root=root))
        else:
            assets = [
                _declared_entry(scope=scope, declared_asset_id=asset_id)
                for asset_id in _strings(scope.get("asset_ids"))
            ]
            catalog.extend(_fit_declared_count(scope, assets))
    return catalog


def _identity_components(values: Iterable[Any]) -> tuple[set[str], set[str]]:
    present_values = [value for value in values if value]
    slugs = {_slug(value) for value in present_values}
    tokens = set().union(*(_identity_tokens(value) for value in present_values))
    return slugs, tokens


def _build_live_identity_indexes(
    live_rows: Sequence[Mapping[str, Any]],
) -> tuple[dict[str, list[int]], dict[str, list[int]]]:
    slug_index: defaultdict[str, list[int]] = defaultdict(list)
    token_index: defaultdict[str, list[int]] = defaultdict(list)
    for index, live in enumerate(live_rows):
        slugs, tokens = _identity_components(
            (live.get("name"), live.get("asset_key"), live.get("source_repo"))
        )
        for slug in slugs:
            slug_index[slug].append(index)
        for token in tokens:
            token_index[token].append(index)
    return dict(slug_index), dict(token_index)


def _best_live_match_index(
    declared: Mapping[str, Any],
    *,
    slug_index: Mapping[str, Sequence[int]],
    token_index: Mapping[str, Sequence[int]],
    used_live_indexes: set[int],
) -> int | None:
    declared_slugs, declared_tokens = _identity_components(
        (
            declared.get("declared_asset_id"),
            declared.get("display_name"),
            declared.get("canonical_id"),
        )
    )
    exact_candidates = {
        index
        for slug in declared_slugs
        for index in slug_index.get(slug, ())
        if index not in used_live_indexes
    }
    if exact_candidates:
        return min(exact_candidates)

    overlap_counts: Counter[int] = Counter()
    for token in declared_tokens:
        overlap_counts.update(
            index
            for index in token_index.get(token, ())
            if index not in used_live_indexes
        )
    if not overlap_counts:
        return None

    # Scores cap at six overlapping tokens; original live order breaks ties.
    return min(
        overlap_counts,
        key=lambda index: (-min(overlap_counts[index], 6), index),
    )


def _is_fresh(value: Any, *, now: datetime, slo_seconds: int) -> bool:
    age = _age_seconds(value, now=now)
    return age is not None and age <= slo_seconds


def _coverage_stage(
    stage_id: str,
    live: Mapping[str, Any] | None,
    *,
    now: datetime,
    run_fresh: bool,
) -> dict[str, Any]:
    if live is None:
        return {
            "status": "missing",
            "evidence_refs": [],
            "reason": "runtime_asset_not_observed",
        }
    coverage = _dict(live.get("coverage"))
    dimensions = _STAGE_DIMENSIONS[stage_id]
    dimension_rows = [_dict(coverage.get(dimension)) for dimension in dimensions]
    evidence_refs = [
        f"asset-coverage:{live.get('asset_id')}:{dimension}"
        for dimension, row in zip(dimensions, dimension_rows, strict=True)
        if row
    ]
    if not run_fresh or any(
        row
        and not _is_fresh(
            row.get("created_at"), now=now, slo_seconds=FRESHNESS_SLO_SECONDS
        )
        for row in dimension_rows
    ):
        return {
            "status": "stale",
            "evidence_refs": evidence_refs,
            "reason": "coverage_receipt_outside_freshness_slo",
        }
    statuses = [str(row.get("coverage_status") or "unknown") for row in dimension_rows]
    if statuses and all(status == "green" for status in statuses):
        return {
            "status": "covered",
            "evidence_refs": evidence_refs,
            "reason": "coverage_dimensions_green",
        }
    return {
        "status": "missing",
        "evidence_refs": evidence_refs,
        "reason": "coverage_dimension_not_green",
    }


def _backup_stage(
    live: Mapping[str, Any] | None,
    *,
    now: datetime,
) -> dict[str, Any]:
    if live is None:
        return {
            "status": "missing",
            "evidence_refs": [],
            "reason": "runtime_asset_not_observed",
        }
    row = _dict(_dict(live.get("compliance")).get("backup_tested"))
    evidence_refs = (
        [f"asset-compliance:{live.get('asset_id')}:backup_tested"] if row else []
    )
    if row and not _is_fresh(
        row.get("detected_at"),
        now=now,
        slo_seconds=RECONCILIATION_FRESHNESS_SLO_SECONDS,
    ):
        return {
            "status": "stale",
            "evidence_refs": evidence_refs,
            "reason": "backup_test_receipt_outside_freshness_slo",
        }
    if str(row.get("status") or "") == "compliant":
        return {
            "status": "covered",
            "evidence_refs": evidence_refs,
            "reason": "backup_test_compliant",
        }
    return {
        "status": "missing",
        "evidence_refs": evidence_refs,
        "reason": "backup_test_not_compliant",
    }


def _build_stages(
    asset: Mapping[str, Any],
    live: Mapping[str, Any] | None,
    *,
    now: datetime,
    run_fresh: bool,
) -> dict[str, dict[str, Any]]:
    source_refs = _strings(asset.get("source_truth_refs"))
    last_seen = live.get("last_seen_at") if live else None
    runtime_fresh = live is not None and _is_fresh(
        last_seen,
        now=now,
        slo_seconds=FRESHNESS_SLO_SECONDS,
    )
    stages = {
        "source_truth": {
            "status": "covered" if source_refs else "missing",
            "evidence_refs": source_refs,
            "reason": "committed_source_refs_present"
            if source_refs
            else "source_ref_missing",
        },
        "runtime_identity": {
            "status": "covered" if runtime_fresh else "stale" if live else "missing",
            "evidence_refs": [f"asset-inventory:{live.get('asset_id')}"]
            if live
            else [],
            "reason": (
                "live_asset_observed"
                if runtime_fresh
                else "runtime_asset_outside_freshness_slo"
                if live
                else "runtime_asset_not_observed"
            ),
        },
    }
    for stage_id in ("sensor", "policy", "executor", "verifier", "learning_writeback"):
        stages[stage_id] = _coverage_stage(
            stage_id,
            live,
            now=now,
            run_fresh=run_fresh,
        )
    stages["backup_restore"] = _backup_stage(live, now=now)
    return {stage_id: stages[stage_id] for stage_id in STAGE_IDS}


def _public_runtime_identity(
    canonical_id: str,
    live: Mapping[str, Any] | None,
) -> dict[str, Any]:
    if live is None:
        return {
            "identity_ref": f"declared:{canonical_id}",
            "observation_status": "declared_only",
            "asset_inventory_id": None,
            "asset_key_fingerprint": "",
            "asset_type": "declared_asset",
            "environment": "unknown",
            "namespace": "",
            "name": "",
            "last_seen_at": "",
        }
    return {
        "identity_ref": f"asset_inventory:{live.get('asset_id')}",
        "observation_status": "live_observed",
        "asset_inventory_id": int(live.get("asset_id") or 0) or None,
        "asset_key_fingerprint": _fingerprint(
            str(live.get("asset_key") or ""), length=16
        ),
        "asset_type": str(live.get("asset_type") or "unknown"),
        "environment": str(live.get("environment") or "unknown"),
        "namespace": str(live.get("namespace") or ""),
        "name": str(live.get("name") or ""),
        "last_seen_at": _iso(live.get("last_seen_at")),
    }


def _live_only_asset(
    live: Mapping[str, Any],
    scope_labels: Mapping[str, str],
) -> dict[str, Any]:
    scope_id = _LIVE_TYPE_SCOPE.get(str(live.get("asset_type") or ""), "services")
    safe_name = str(live.get("name") or live.get("asset_type") or "discovered")
    key_fingerprint = _fingerprint(str(live.get("asset_key") or safe_name), length=12)
    return {
        "canonical_id": (
            f"awoooi:{scope_id}:discovered-{_slug(safe_name)}-{key_fingerprint}"
        ),
        "scope_id": scope_id,
        "scope_label": scope_labels.get(scope_id, scope_id),
        "declared_asset_id": "",
        "display_name": safe_name,
        "owner_lane": str(
            live.get("owner_team") or _OWNER_LANES.get(scope_id, "platform_governance")
        ),
        "source_truth_refs": [
            str(live.get("source_repo") or f"asset_inventory:{live.get('asset_id')}")
        ],
        "declared": False,
    }


def _receipt_controls(
    receipt: Mapping[str, Any] | None,
    *,
    matrix_fingerprint: str,
    now: datetime,
) -> dict[str, bool]:
    row = _dict(receipt)
    output = _dict(row.get("output"))
    candidate_count = int(output.get("candidate_count") or 0)
    readback_count = int(output.get("repository_readback_count") or 0)
    created_at = row.get("created_at")
    fresh = bool(
        row
        and str(row.get("status") or "") == "success"
        and _is_fresh(
            created_at,
            now=now,
            slo_seconds=RECONCILIATION_FRESHNESS_SLO_SECONDS,
        )
    )
    return {
        "daily_reconciliation_receipt": fresh,
        "reconciliation_matrix_fingerprint_match": bool(
            fresh and str(output.get("matrix_fingerprint") or "") == matrix_fingerprint
        ),
        "drift_work_items_verified": bool(
            fresh
            and output.get("repository_readback_verified") is True
            and candidate_count == readback_count
        ),
    }


def _matrix_fingerprint_row(asset: Mapping[str, Any]) -> dict[str, Any]:
    return {
        "canonical_id": asset["canonical_id"],
        "declared": asset["declared"],
        "runtime_identity": _dict(asset.get("runtime_identity")).get(
            "asset_key_fingerprint"
        ),
        "stages": {
            stage_id: _dict(_dict(asset.get("stages")).get(stage_id)).get("status")
            for stage_id in STAGE_IDS
        },
        "change_types": asset.get("change_types"),
    }


def build_asset_capability_matrix(
    scope_classes: Sequence[Mapping[str, Any]],
    *,
    live_assets: Sequence[Mapping[str, Any]] = (),
    latest_run: Mapping[str, Any] | None = None,
    reconciliation_receipt: Mapping[str, Any] | None = None,
    repo_root: Path | None = None,
    generated_at: datetime | None = None,
    live_readback_status: str = "ready",
    include_live_only_assets: bool = True,
) -> dict[str, Any]:
    """Build a public-safe per-asset matrix from declared and live truth."""

    now = (generated_at or datetime.now(UTC)).astimezone(UTC)
    declared_assets = build_declared_asset_catalog(scope_classes, repo_root=repo_root)
    live_rows = [dict(row) for row in live_assets]
    scope_labels = {
        str(scope.get("id") or ""): str(scope.get("label") or scope.get("id") or "")
        for scope in scope_classes
    }
    latest = _dict(latest_run)
    run_fresh = _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()
    assets: list[dict[str, Any]] = []

    for declared in declared_assets:
        matched_index = _best_live_match_index(
            declared,
            slug_index=slug_index,
            token_index=token_index,
            used_live_indexes=used_live_indexes,
        )
        matched_live = live_rows[matched_index] if matched_index is not None else None
        if matched_index is not None:
            used_live_indexes.add(matched_index)
        stages = _build_stages(
            declared,
            matched_live,
            now=now,
            run_fresh=run_fresh,
        )
        assets.append(
            {
                **declared,
                "runtime_identity": _public_runtime_identity(
                    str(declared["canonical_id"]),
                    matched_live,
                ),
                "stages": stages,
                "missing_stage_ids": [
                    stage_id
                    for stage_id, stage in stages.items()
                    if stage.get("status") == "missing"
                ],
                "stale_stage_ids": [
                    stage_id
                    for stage_id, stage in stages.items()
                    if stage.get("status") == "stale"
                ],
                "change_types": _strings(
                    matched_live.get("change_types") if matched_live else []
                ),
            }
        )

    fingerprint_rows = [_matrix_fingerprint_row(asset) for asset in assets]
    unmatched_live_indexes = [
        index for index in range(len(live_rows)) if index not in used_live_indexes
    ]
    for live_index in unmatched_live_indexes:
        live = live_rows[live_index]
        discovered = _live_only_asset(live, scope_labels)
        stages = _build_stages(discovered, live, now=now, run_fresh=run_fresh)
        live_only_asset = {
            **discovered,
            "runtime_identity": _public_runtime_identity(
                str(discovered["canonical_id"]),
                live,
            ),
            "stages": stages,
            "missing_stage_ids": [
                stage_id
                for stage_id, stage in stages.items()
                if stage.get("status") == "missing"
            ],
            "stale_stage_ids": [
                stage_id
                for stage_id, stage in stages.items()
                if stage.get("status") == "stale"
            ],
            "change_types": list(
                dict.fromkeys(["asset_added", *_strings(live.get("change_types"))])
            ),
        }
        fingerprint_rows.append(_matrix_fingerprint_row(live_only_asset))
        if include_live_only_assets:
            assets.append(live_only_asset)

    assets.sort(key=lambda row: (str(row["scope_id"]), str(row["canonical_id"])))
    fingerprint_rows.sort(key=lambda row: str(row["canonical_id"]))
    matrix_fingerprint = _fingerprint(fingerprint_rows)
    canonical_ids = [str(asset.get("canonical_id") or "") for asset in assets]
    expected_scope_counts = {
        str(scope.get("id") or ""): int(scope.get("asset_count") or 0)
        for scope in scope_classes
    }
    declared_scope_counts = Counter(
        str(asset.get("scope_id") or "")
        for asset in assets
        if asset.get("declared") is True
    )
    controls = {
        "canonical_id_per_asset": bool(
            assets
            and all(canonical_ids)
            and len(canonical_ids) == len(set(canonical_ids))
        ),
        "owner_lane_per_asset": bool(
            assets and all(str(asset.get("owner_lane") or "") for asset in assets)
        ),
        "source_truth_per_asset": bool(
            assets and all(_strings(asset.get("source_truth_refs")) for asset in assets)
        ),
        "runtime_identity_per_asset": bool(
            assets
            and all(
                str(_dict(asset.get("runtime_identity")).get("identity_ref") or "")
                for asset in assets
            )
        ),
        "per_asset_stage_status": bool(
            assets
            and all(
                set(_dict(asset.get("stages"))) == set(STAGE_IDS)
                and all(
                    _dict(_dict(asset.get("stages")).get(stage_id)).get("status")
                    in {"covered", "missing", "stale"}
                    for stage_id in STAGE_IDS
                )
                for asset in assets
            )
        ),
        "full_scope_class_coverage": bool(
            expected_scope_counts
            and all(
                declared_scope_counts.get(scope_id, 0) == expected
                for scope_id, expected in expected_scope_counts.items()
            )
        ),
    }
    controls.update(
        _receipt_controls(
            reconciliation_receipt,
            matrix_fingerprint=matrix_fingerprint,
            now=now,
        )
    )
    stage_counts = Counter(
        str(stage.get("status") or "missing")
        for asset in assets
        for stage in _dict(asset.get("stages")).values()
        if isinstance(stage, dict)
    )
    gap_assets = [
        asset
        for asset in assets
        if asset.get("missing_stage_ids") or asset.get("stale_stage_ids")
    ]
    live_observed_count = sum(
        1
        for asset in assets
        if _dict(asset.get("runtime_identity")).get("observation_status")
        == "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 {
        "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,
        "freshness_slo_seconds": FRESHNESS_SLO_SECONDS,
        "reconciliation_freshness_slo_seconds": RECONCILIATION_FRESHNESS_SLO_SECONDS,
        "latest_discovery_run": {
            "run_id": str(latest.get("run_id") or ""),
            "ended_at": _iso(latest.get("ended_at")),
            "fresh": run_fresh,
            "status": str(latest.get("status") or "unavailable"),
        },
        "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),
            "live_inventory_asset_count": len(live_rows),
            "matrix_asset_count": len(assets),
            "reconciled_asset_count": len(declared_assets)
            + len(unmatched_live_indexes),
            "live_only_assets_embedded": include_live_only_assets,
            "embedded_live_only_asset_count": (
                len(unmatched_live_indexes) if include_live_only_assets else 0
            ),
            "live_observed_asset_count": live_observed_count,
            "declared_only_asset_count": len(declared_assets)
            - sum(
                1
                for asset in assets
                if asset.get("declared")
                and _dict(asset.get("runtime_identity")).get("observation_status")
                == "live_observed"
            ),
            "discovered_not_declared_count": len(unmatched_live_indexes),
            "gap_asset_count": len(gap_assets),
            "covered_stage_cell_count": stage_counts["covered"],
            "missing_stage_cell_count": stage_counts["missing"],
            "stale_stage_cell_count": stage_counts["stale"],
            "stage_cell_count": total_stage_cells,
            "stage_coverage_percent": (
                round(stage_counts["covered"] / total_stage_cells * 100)
                if total_stage_cells
                else 0
            ),
            "runtime_observation_percent": (
                round(live_observed_count / len(assets) * 100) if assets else 0
            ),
            "control_pass_count": sum(controls.values()),
            "required_control_count": len(REQUIRED_CONTROL_IDS),
            "reconciliation_candidate_count": int(
                receipt_output.get("candidate_count") or 0
            ),
            "reconciliation_work_item_readback_count": int(
                receipt_output.get("repository_readback_count") or 0
            ),
        },
        "scope_rollups": [
            {
                "scope_id": scope_id,
                "scope_label": scope_labels.get(scope_id, scope_id),
                "expected_declared_asset_count": expected_scope_counts.get(scope_id, 0),
                "declared_asset_count": sum(
                    1
                    for asset in assets
                    if asset.get("scope_id") == scope_id
                    and asset.get("declared") is True
                ),
                "matrix_asset_count": sum(
                    1 for asset in assets if asset.get("scope_id") == scope_id
                ),
                "gap_asset_count": sum(
                    1 for asset in gap_assets if asset.get("scope_id") == scope_id
                ),
            }
            for scope_id in expected_scope_counts
        ],
        "assets": assets,
        "reconciliation_receipt": {
            "op_id": str(receipt.get("op_id") or ""),
            "status": str(receipt.get("status") or "unavailable"),
            "created_at": _iso(receipt.get("created_at")),
            "trace_id": str(_dict(receipt.get("input")).get("trace_id") or ""),
            "run_id": str(_dict(receipt.get("input")).get("automation_run_id") or ""),
            "work_item_id": str(_dict(receipt.get("input")).get("work_item_id") or ""),
            "repository_readback_verified": receipt_output.get(
                "repository_readback_verified"
            )
            is True,
        },
        "operation_boundaries": {
            "read_only_api": True,
            "raw_asset_key_exposed": False,
            "raw_host_exposed": False,
            "secret_value_read": False,
            "github_used": False,
            "host_write_performed": False,
        },
    }


def build_reconciliation_candidates(matrix: Mapping[str, Any]) -> list[dict[str, Any]]:
    """Return deterministic work-item candidates for gaps and asset drift."""

    candidates: list[dict[str, Any]] = []
    matrix_fingerprint = str(matrix.get("matrix_fingerprint") or "")
    trace_id = f"asset-capability:{matrix_fingerprint[:16]}"
    automation_run_id = str(
        _dict(matrix.get("latest_discovery_run")).get("run_id")
        or matrix_fingerprint[:32]
    )
    for raw_asset in _list(matrix.get("assets")):
        asset = _dict(raw_asset)
        missing = _strings(asset.get("missing_stage_ids"))
        stale = _strings(asset.get("stale_stage_ids"))
        changes = _strings(asset.get("change_types"))
        if not missing and not stale and not changes:
            continue
        candidate_kind = (
            "asset_added"
            if "asset_added" in changes or asset.get("declared") is False
            else "asset_removed"
            if "asset_removed" in changes
            else "asset_drift"
            if changes
            else "capability_gap"
        )
        fingerprint = _fingerprint(
            {
                "canonical_id": asset.get("canonical_id"),
                "missing": missing,
                "stale": stale,
                "changes": changes,
                "kind": candidate_kind,
            },
            length=24,
        )
        candidates.append(
            {
                "schema_version": RECONCILIATION_SCHEMA_VERSION,
                "candidate_fingerprint": fingerprint,
                "work_item_id": f"AIA-P0-006-WI-{fingerprint[:12]}",
                "trace_id": trace_id,
                "automation_run_id": automation_run_id,
                "matrix_fingerprint": matrix_fingerprint,
                "canonical_asset_id": str(asset.get("canonical_id") or ""),
                "asset_inventory_id": _dict(asset.get("runtime_identity")).get(
                    "asset_inventory_id"
                ),
                "scope_id": str(asset.get("scope_id") or ""),
                "candidate_kind": candidate_kind,
                "missing_stage_ids": missing,
                "stale_stage_ids": stale,
                "change_types": changes,
                "risk_level": "low",
                "controlled_apply_allowed": True,
                "idempotency_key": f"asset-capability:{fingerprint}",
                "post_verifier": "automation_operation_log_row_readback",
            }
        )
    candidates.sort(key=lambda row: str(row["canonical_asset_id"]))
    return candidates


async def _load_live_rows(
    project_id: str,
) -> tuple[
    dict[str, Any],
    list[dict[str, Any]],
    dict[str, Any],
]:
    from sqlalchemy import text

    from src.db.base import get_db_context

    async with get_db_context(project_id) as db:
        await db.execute(
            text("SELECT set_config('statement_timeout', :statement_timeout, true)"),
            {
                "statement_timeout": (
                    f"{LIVE_READBACK_DB_STATEMENT_TIMEOUT_MILLISECONDS}ms"
                )
            },
        )
        latest_result = await db.execute(
            text(
                """
                SELECT run_id::text AS run_id, status, ended_at
                FROM asset_discovery_run
                WHERE status IN ('success', 'partial')
                ORDER BY ended_at DESC NULLS LAST
                LIMIT 1
                """
            )
        )
        latest_row = latest_result.mappings().first()
        latest_run = dict(latest_row) if latest_row else {}
        run_id = str(latest_run.get("run_id") or "")

        assets_result = await db.execute(
            text(
                """
                SELECT
                    ai.asset_id,
                    ai.asset_key,
                    ai.asset_type,
                    ai.environment,
                    ai.namespace,
                    ai.name,
                    ai.owner_team,
                    ai.source_repo,
                    ai.source_commit_sha,
                    ai.lifecycle_state,
                    ai.last_seen_at,
                    cs.dimension,
                    cs.coverage_status,
                    cs.created_at AS coverage_created_at
                FROM asset_inventory ai
                LEFT 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'
                ORDER BY ai.asset_id, cs.dimension
                """
            ),
            {"run_id": run_id},
        )
        assets_by_id: dict[int, dict[str, Any]] = {}
        for row in assets_result.mappings().all():
            item = dict(row)
            asset_id = int(item["asset_id"])
            asset = assets_by_id.setdefault(
                asset_id,
                {
                    key: item.get(key)
                    for key in (
                        "asset_id",
                        "asset_key",
                        "asset_type",
                        "environment",
                        "namespace",
                        "name",
                        "owner_team",
                        "source_repo",
                        "source_commit_sha",
                        "lifecycle_state",
                        "last_seen_at",
                    )
                }
                | {"coverage": {}, "compliance": {}, "change_types": []},
            )
            dimension = str(item.get("dimension") or "")
            if dimension:
                asset["coverage"][dimension] = {
                    "coverage_status": str(item.get("coverage_status") or "unknown"),
                    "created_at": item.get("coverage_created_at"),
                }

        asset_ids = list(assets_by_id) or [0]
        compliance_result = await db.execute(
            text(
                """
                SELECT DISTINCT ON (asset_id, dimension)
                    asset_id, dimension, status, detected_at
                FROM asset_compliance_snapshot
                WHERE asset_id = ANY(:asset_ids)
                ORDER BY asset_id, dimension, detected_at DESC
                """
            ),
            {"asset_ids": asset_ids},
        )
        for row in compliance_result.mappings().all():
            item = dict(row)
            asset = assets_by_id.get(int(item["asset_id"]))
            if asset is not None:
                asset["compliance"][str(item["dimension"])] = {
                    "status": str(item.get("status") or "unknown"),
                    "detected_at": item.get("detected_at"),
                }

        changes_result = await db.execute(
            text(
                """
                SELECT asset_id, change_type
                FROM asset_change_event
                WHERE detected_at >= NOW() - INTERVAL '36 hours'
                  AND asset_id IS NOT NULL
                  AND asset_id = ANY(:asset_ids)
                ORDER BY detected_at DESC
                """
            ),
            {"asset_ids": asset_ids},
        )
        for row in changes_result.mappings().all():
            item = dict(row)
            asset = assets_by_id.get(int(item["asset_id"]))
            change_type = str(item.get("change_type") or "")
            if asset is not None and change_type:
                asset["change_types"] = list(
                    dict.fromkeys([*asset["change_types"], change_type])
                )

        receipt_result = await db.execute(
            text(
                """
                SELECT op_id::text AS op_id, status, input, output, created_at
                FROM automation_operation_log
                WHERE actor = :actor
                  AND COALESCE(input ->> 'semantic_operation_type', operation_type)
                      = :semantic_operation_type
                ORDER BY created_at DESC
                LIMIT 1
                """
            ),
            {
                "actor": RECONCILIATION_ACTOR,
                "semantic_operation_type": RECONCILIATION_SUMMARY_OPERATION,
            },
        )
        receipt_row = receipt_result.mappings().first()
        receipt = dict(receipt_row) if receipt_row else {}

    return latest_run, list(assets_by_id.values()), receipt


async def build_asset_capability_matrix_with_live_readback(
    *,
    project_id: str = DEFAULT_PROJECT_ID,
    repo_root: Path | None = None,
    timeout_seconds: float = LIVE_READBACK_TIMEOUT_SECONDS,
    include_live_only_assets: bool = False,
) -> dict[str, Any]:
    """Build a bounded matrix from live DB truth, failing closed to declared scope."""

    from src.services.awoooi_priority_work_order_readback import (
        get_ai_automation_program_scope_classes,
    )

    scope_classes = get_ai_automation_program_scope_classes()
    bounded_timeout_seconds = max(0.001, float(timeout_seconds))

    async def _load_and_build() -> dict[str, Any]:
        latest_run, live_assets, receipt = await _load_live_rows(project_id)
        return await asyncio.to_thread(
            build_asset_capability_matrix,
            scope_classes,
            live_assets=live_assets,
            latest_run=latest_run,
            reconciliation_receipt=receipt,
            repo_root=repo_root,
            live_readback_status="ready",
            include_live_only_assets=include_live_only_assets,
        )

    try:
        payload = await asyncio.wait_for(
            _load_and_build(),
            timeout=bounded_timeout_seconds,
        )
    except Exception as exc:
        logger.warning(
            "asset_capability_matrix_live_readback_degraded",
            error_type=type(exc).__name__,
            timeout_seconds=bounded_timeout_seconds,
        )
        payload = build_asset_capability_matrix(
            scope_classes,
            repo_root=repo_root,
            live_readback_status="degraded",
            include_live_only_assets=include_live_only_assets,
        )
        payload["live_readback_error_type"] = type(exc).__name__
        payload["live_readback_timeout_seconds"] = bounded_timeout_seconds
        return payload
    payload["live_readback_timeout_seconds"] = bounded_timeout_seconds
    return payload