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 6bd56a430..81233f1af 100644 --- a/apps/api/src/services/ai_automation_asset_capability_matrix.py +++ b/apps/api/src/services/ai_automation_asset_capability_matrix.py @@ -111,6 +111,24 @@ _STAGE_DIMENSIONS = { "learning_writeback": ("auto_km_creation", "auto_playbook"), } +# These tokens describe the AWOOOI deployment namespace or public DNS suffix, +# not an individual runtime asset. Letting them create candidates causes every +# declared row to match an arbitrary AWOOOI inventory row once a real identity +# has already been consumed. +_NON_DISCRIMINATING_IDENTITY_TOKENS = { + "and", + "app", + "apps", + "com", + "default", + "dev", + "prod", + "production", + "staging", + "wooo", + "work", +} + def _repo_root() -> Path: source_path = Path(__file__).resolve() @@ -365,6 +383,7 @@ 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)) + tokens.difference_update(_NON_DISCRIMINATING_IDENTITY_TOKENS) return slugs, tokens @@ -374,9 +393,7 @@ def _build_live_identity_indexes( 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")) - ) + slugs, tokens = _identity_components((live.get("name"), live.get("asset_key"))) for slug in slugs: slug_index[slug].append(index) for token in tokens: @@ -387,41 +404,87 @@ def _build_live_identity_indexes( def _best_live_match_index( declared: Mapping[str, Any], *, + live_rows: Sequence[Mapping[str, Any]], slug_index: Mapping[str, Sequence[int]], token_index: Mapping[str, Sequence[int]], used_live_indexes: set[int], -) -> int | None: +) -> tuple[int | None, str, float]: + declared_scope_id = str(declared.get("scope_id") or "") + + def _eligible(index: int) -> bool: + if index in used_live_indexes: + return False + live_type = str(live_rows[index].get("asset_type") or "") + return _LIVE_TYPE_SCOPE.get(live_type) == declared_scope_id + 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 _eligible(index) } - if exact_candidates: - return min(exact_candidates) + if len(exact_candidates) == 1: + return next(iter(exact_candidates)), "exact_identity_slug", 1.0 + if len(exact_candidates) > 1: + return None, "ambiguous_exact_identity", 0.0 + + if not declared_tokens: + return None, "declared_identity_tokens_missing", 0.0 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 + index for index in token_index.get(token, ()) if _eligible(index) ) if not overlap_counts: - return None + return None, "no_compatible_identity_evidence", 0.0 - # Scores cap at six overlapping tokens; original live order breaks ties. - return min( - overlap_counts, - key=lambda index: (-min(overlap_counts[index], 6), index), + minimum_overlap = min(2, len(declared_tokens)) + scored_candidates: dict[int, tuple[int, int]] = {} + live_token_counts: dict[int, int] = {} + for index, overlap_count in overlap_counts.items(): + if overlap_count < minimum_overlap: + continue + _live_slugs, live_tokens = _identity_components( + (live_rows[index].get("name"), live_rows[index].get("asset_key")) + ) + if not live_tokens: + continue + declared_coverage = overlap_count / len(declared_tokens) + if declared_coverage < 0.5: + continue + live_token_counts[index] = len(live_tokens) + scored_candidates[index] = ( + overlap_count, + -(len(live_tokens) - overlap_count), + ) + if not scored_candidates: + return None, "identity_evidence_below_threshold", 0.0 + + best_score = max(scored_candidates.values()) + best_candidates = [ + index for index, score in scored_candidates.items() if score == best_score + ] + if len(best_candidates) != 1: + return None, "ambiguous_scored_identity", 0.0 + + best_index = best_candidates[0] + overlap_count = overlap_counts[best_index] + confidence = round( + ( + overlap_count / len(declared_tokens) + + overlap_count / live_token_counts[best_index] + ) + / 2, + 3, ) + return best_index, "meaningful_identity_token_overlap", confidence def _is_fresh(value: Any, *, now: datetime, slo_seconds: int) -> bool: @@ -564,11 +627,17 @@ def _build_stages( def _public_runtime_identity( canonical_id: str, live: Mapping[str, Any] | None, + *, + match_basis: str, + match_confidence: float, ) -> dict[str, Any]: if live is None: return { "identity_ref": f"declared:{canonical_id}", "observation_status": "declared_only", + "match_status": "unresolved", + "match_basis": match_basis, + "match_confidence": 0.0, "asset_inventory_id": None, "asset_key_fingerprint": "", "asset_type": "declared_asset", @@ -580,6 +649,9 @@ def _public_runtime_identity( return { "identity_ref": f"asset_inventory:{live.get('asset_id')}", "observation_status": "live_observed", + "match_status": "verified", + "match_basis": match_basis, + "match_confidence": match_confidence, "asset_inventory_id": int(live.get("asset_id") or 0) or None, "asset_key_fingerprint": _fingerprint( str(live.get("asset_key") or ""), length=16 @@ -627,9 +699,7 @@ 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 - ) + verifier_readback_count = int(output.get("work_item_verifier_readback_count") or 0) created_at = row.get("created_at") fresh = bool( row @@ -656,12 +726,13 @@ def _receipt_controls( def _matrix_fingerprint_row(asset: Mapping[str, Any]) -> dict[str, Any]: + runtime_identity = _dict(asset.get("runtime_identity")) return { "canonical_id": asset["canonical_id"], "declared": asset["declared"], - "runtime_identity": _dict(asset.get("runtime_identity")).get( - "asset_key_fingerprint" - ), + "runtime_identity": runtime_identity.get("asset_key_fingerprint"), + "runtime_identity_match_status": runtime_identity.get("match_status"), + "runtime_identity_match_basis": runtime_identity.get("match_basis"), "stages": { stage_id: _dict(_dict(asset.get("stages")).get(stage_id)).get("status") for stage_id in STAGE_IDS @@ -701,8 +772,9 @@ def build_asset_capability_matrix( assets: list[dict[str, Any]] = [] for declared in declared_assets: - matched_index = _best_live_match_index( + matched_index, match_basis, match_confidence = _best_live_match_index( declared, + live_rows=live_rows, slug_index=slug_index, token_index=token_index, used_live_indexes=used_live_indexes, @@ -722,6 +794,8 @@ def build_asset_capability_matrix( "runtime_identity": _public_runtime_identity( str(declared["canonical_id"]), matched_live, + match_basis=match_basis, + match_confidence=match_confidence, ), "stages": stages, "missing_stage_ids": [ @@ -753,6 +827,8 @@ def build_asset_capability_matrix( "runtime_identity": _public_runtime_identity( str(discovered["canonical_id"]), live, + match_basis="live_inventory_discovery", + match_confidence=1.0, ), "stages": stages, "missing_stage_ids": [ @@ -802,6 +878,8 @@ def build_asset_capability_matrix( assets and all( str(_dict(asset.get("runtime_identity")).get("identity_ref") or "") + and _dict(asset.get("runtime_identity")).get("match_status") + == "verified" for asset in assets ) ), 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 8235dbb42..4642c75e2 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/apps/api/tests/test_ai_automation_asset_capability_matrix.py @@ -154,7 +154,7 @@ def test_public_projection_counts_live_only_assets_without_embedding_them() -> N ) -def test_indexed_matching_preserves_score_cap_and_original_live_order() -> None: +def test_indexed_match_is_order_independent_and_uses_strongest_candidate() -> None: now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC) scope = [ { @@ -165,7 +165,7 @@ def test_indexed_matching_preserves_score_cap_and_original_live_order() -> None: "inventory_sources": ["service-inventory.json"], } ] - live_assets = [ + candidates = [ { "asset_id": 1, "asset_key": "candidate:first", @@ -182,6 +182,151 @@ def test_indexed_matching_preserves_score_cap_and_original_live_order() -> None: }, ] + for live_assets in (candidates, list(reversed(candidates))): + matrix = build_asset_capability_matrix( + scope, + live_assets=live_assets, + repo_root=_REPO_ROOT, + generated_at=now, + include_live_only_assets=False, + ) + + runtime_identity = matrix["assets"][0]["runtime_identity"] + assert runtime_identity["asset_inventory_id"] == 2 + assert runtime_identity["match_status"] == "verified" + assert runtime_identity["match_basis"] == "meaningful_identity_token_overlap" + + +def test_common_repo_token_and_cross_scope_type_cannot_fabricate_match() -> None: + now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC) + scope = [ + { + "id": "observability_and_security", + "label": "Observability", + "asset_count": 1, + "asset_ids": ["sentry"], + "inventory_sources": ["observability-inventory.json"], + } + ] + live_assets = [ + { + "asset_id": 1, + "asset_key": "package:otel-collector:otel_collector", + "asset_type": "package", + "name": "otel-collector/otel_collector", + "source_repo": "wooo/awoooi", + "last_seen_at": now, + } + ] + + initial = build_asset_capability_matrix( + scope, + live_assets=live_assets, + latest_run={"run_id": "run-1", "status": "success", "ended_at": now}, + repo_root=_REPO_ROOT, + generated_at=now, + include_live_only_assets=False, + ) + matrix = build_asset_capability_matrix( + scope, + live_assets=live_assets, + 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, + include_live_only_assets=False, + ) + + runtime_identity = matrix["assets"][0]["runtime_identity"] + assert runtime_identity["observation_status"] == "declared_only" + assert runtime_identity["match_status"] == "unresolved" + assert runtime_identity["match_basis"] == "no_compatible_identity_evidence" + assert runtime_identity["asset_inventory_id"] is None + assert matrix["summary"]["live_observed_asset_count"] == 0 + assert matrix["summary"]["declared_only_asset_count"] == 1 + assert matrix["controls"]["runtime_identity_per_asset"] is False + assert matrix["closed"] is False + + +def test_typed_website_match_ignores_unrelated_container_regardless_of_order() -> None: + now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC) + scope = [ + { + "id": "websites", + "label": "Websites", + "asset_count": 1, + "asset_ids": ["sentry.wooo.work"], + "inventory_sources": ["website-inventory.json"], + } + ] + candidates = [ + { + "asset_id": 1, + "asset_key": "container:awoooi-prod:km-vectorize", + "asset_type": "container", + "name": "km-vectorize-29732820-bx479", + "source_repo": "wooo/awoooi", + "last_seen_at": now, + }, + { + "asset_id": 2, + "asset_key": "website:sentry.wooo.work", + "asset_type": "website", + "name": "sentry.wooo.work", + "last_seen_at": now, + }, + ] + + for live_assets in (candidates, list(reversed(candidates))): + matrix = build_asset_capability_matrix( + scope, + live_assets=live_assets, + repo_root=_REPO_ROOT, + generated_at=now, + include_live_only_assets=False, + ) + + runtime_identity = matrix["assets"][0]["runtime_identity"] + assert runtime_identity["asset_inventory_id"] == 2 + assert runtime_identity["asset_type"] == "website" + assert runtime_identity["name"] == "sentry.wooo.work" + assert runtime_identity["match_basis"] == "exact_identity_slug" + assert runtime_identity["match_confidence"] == 1.0 + + +def test_ambiguous_exact_live_identity_fails_closed() -> None: + now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC) + scope = [ + { + "id": "websites", + "label": "Websites", + "asset_count": 1, + "asset_ids": ["sentry.wooo.work"], + "inventory_sources": ["website-inventory.json"], + } + ] + live_assets = [ + { + "asset_id": index, + "asset_key": f"website:{index}:sentry.wooo.work", + "asset_type": "website", + "name": "sentry.wooo.work", + "last_seen_at": now, + } + for index in (1, 2) + ] + matrix = build_asset_capability_matrix( scope, live_assets=live_assets, @@ -190,7 +335,47 @@ def test_indexed_matching_preserves_score_cap_and_original_live_order() -> None: include_live_only_assets=False, ) - assert matrix["assets"][0]["runtime_identity"]["asset_inventory_id"] == 1 + runtime_identity = matrix["assets"][0]["runtime_identity"] + assert runtime_identity["observation_status"] == "declared_only" + assert runtime_identity["match_status"] == "unresolved" + assert runtime_identity["match_basis"] == "ambiguous_exact_identity" + assert matrix["controls"]["runtime_identity_per_asset"] is False + + +def test_declared_hosts_remain_unresolved_when_inventory_only_has_k8s_assets() -> None: + now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC) + host_scope = [ + scope + for scope in get_ai_automation_program_scope_classes() + if scope["id"] == "hosts" + ] + live_assets = [ + { + "asset_id": index, + "asset_key": f"k8s:deployment:awoooi-api:{index}", + "asset_type": "k8s_workload", + "name": f"awoooi-api-{index}", + "source_repo": "wooo/awoooi", + "last_seen_at": now, + } + for index in range(1, 8) + ] + + matrix = build_asset_capability_matrix( + host_scope, + live_assets=live_assets, + repo_root=_REPO_ROOT, + generated_at=now, + include_live_only_assets=False, + ) + + assert matrix["summary"]["declared_asset_count"] == 7 + assert matrix["summary"]["live_observed_asset_count"] == 0 + assert matrix["summary"]["declared_only_asset_count"] == 7 + assert { + asset["runtime_identity"]["match_status"] for asset in matrix["assets"] + } == {"unresolved"} + assert matrix["controls"]["runtime_identity_per_asset"] is False def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> None: @@ -322,10 +507,25 @@ def test_reconciliation_candidates_are_stable_and_public_safe() -> None: def test_reconciliation_summary_requires_repository_readback() -> None: + now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC) matrix = build_asset_capability_matrix( _single_host_scope(), - latest_run={"run_id": "discovery-run-1", "status": "success"}, + live_assets=[ + { + "asset_id": 42, + "asset_key": "host:99", + "asset_type": "host", + "name": "99", + "last_seen_at": now, + } + ], + latest_run={ + "run_id": "discovery-run-1", + "status": "success", + "ended_at": now, + }, repo_root=_REPO_ROOT, + generated_at=now, ) candidates = build_reconciliation_candidates(matrix) @@ -808,9 +1008,7 @@ class _FakeReconciliationDb: if "UPDATE automation_operation_log" in sql: assert isinstance(params, dict) requested = set(params["candidate_fingerprints"]) - self.verified_fingerprints.update( - requested & self.persisted_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) @@ -825,9 +1023,7 @@ class _FakeReconciliationDb: ) if "RETURNING op_id::text" in sql: assert isinstance(params, dict) - self.summary_outputs.append( - json.loads(str(params["output"])) - ) + self.summary_outputs.append(json.loads(str(params["output"]))) return _FakeReconciliationResult(scalar="summary-op-1") raise AssertionError(f"unexpected SQL: {sql}") @@ -853,13 +1049,8 @@ def test_reconciliation_persists_bounded_batches_and_reuses_verified_summary( ) 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] - }, + "latest_discovery_run": {"run_id": "00000000-0000-0000-0000-000000000001"}, + "controls": {control_id: True for control_id in REQUIRED_CONTROL_IDS[:6]}, } candidates = [ {