wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

fix(assets): fail closed on ambiguous runtime identity

Browse Source
This commit is contained in:
ogt
2026-07-15 14:17:24 +08:00
parent c56dee39ed
commit 94f8624b02
2 changed files with 309 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
apps/api/src/services/ai_automation_asset_capability_matrix.py
View File

@@ -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
)
),