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

Merge remote-tracking branch 'origin/main' into codex/telegram-routing-registry-20260714

Browse Source
This commit is contained in:
ogt
2026-07-15 05:14:52 +08:00
parent 2bb6d8eede 228d474534
commit d683ec5359
7 changed files with 282 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
apps/api/src/jobs/asset_capability_reconciliation_job.py
View File

@@ -216,7 +216,8 @@ async def reconcile_once(
started = time.monotonic()
matrix = await build_asset_capability_matrix_with_live_readback(
project_id=project_id
project_id=project_id,
include_live_only_assets=True,
)
if matrix.get("live_readback_status") != "ready":
logger.warning(

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

@@ -12,7 +12,7 @@ import asyncio
import hashlib
import json
import re
from collections import Counter
from collections import Counter, defaultdict
from collections.abc import Iterable, Mapping, Sequence
from datetime import UTC, datetime
from pathlib import Path
@@ -31,6 +31,7 @@ RECONCILIATION_WORK_ITEM_OPERATION = "asset_capability_reconciliation_work_item_
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",
@@ -354,25 +355,67 @@ def build_declared_asset_catalog(
return catalog
def _matching_score(declared: Mapping[str, Any], live: Mapping[str, Any]) -> int:
declared_values = (
declared.get("declared_asset_id"),
declared.get("display_name"),
declared.get("canonical_id"),
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"),
)
)
live_values = (live.get("name"), live.get("asset_key"), live.get("source_repo"))
declared_slugs = {_slug(value) for value in declared_values if value}
live_slugs = {_slug(value) for value in live_values if value}
if declared_slugs & live_slugs:
return 100
declared_tokens = set().union(
*(_identity_tokens(value) for value in declared_values)
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),
)
live_tokens = set().union(*(_identity_tokens(value) for value in live_values))
overlap = declared_tokens & live_tokens
if overlap:
return min(80, 20 + (len(overlap) * 10))
return 0
def _is_fresh(value: Any, *, now: datetime, slo_seconds: int) -> bool:
@@ -601,6 +644,21 @@ def _receipt_controls(
}
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]],
*,
@@ -610,6 +668,7 @@ def build_asset_capability_matrix(
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."""
@@ -626,19 +685,20 @@ def build_asset_capability_matrix(
now=now,
slo_seconds=FRESHNESS_SLO_SECONDS,
)
used_live_ids: set[int] = set()
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:
candidates = [
(live, _matching_score(declared, live))
for live in live_rows
if int(live.get("asset_id") or 0) not in used_live_ids
]
live, score = max(candidates, key=lambda item: item[1], default=(None, 0))
matched_live = live if score >= 30 else None
if matched_live is not None:
used_live_ids.add(int(matched_live.get("asset_id") or 0))
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,
@@ -669,52 +729,41 @@ def build_asset_capability_matrix(
}
)
for live in live_rows:
live_id = int(live.get("asset_id") or 0)
if live_id in used_live_ids:
continue
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)
assets.append(
{
**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"))])
),
}
)
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 = [
{
"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"),
}
for asset in assets
]
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 = {
@@ -819,6 +868,12 @@ def build_asset_capability_matrix(
"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(
@@ -828,9 +883,7 @@ def build_asset_capability_matrix(
and _dict(asset.get("runtime_identity")).get("observation_status")
== "live_observed"
),
"discovered_not_declared_count": sum(
1 for asset in assets if asset.get("declared") is False
),
"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"],
@@ -972,6 +1025,14 @@ async def _load_live_rows(
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(
"""
@@ -1118,8 +1179,9 @@ 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 the matrix from live DB truth, failing closed to declared scope."""
"""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,
@@ -1127,9 +1189,23 @@ async def build_asset_capability_matrix_with_live_readback(
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:
latest_run, live_assets, receipt = await asyncio.wait_for(
_load_live_rows(project_id),
payload = await asyncio.wait_for(
_load_and_build(),
timeout=bounded_timeout_seconds,
)
except Exception as exc:
@@ -1142,17 +1218,10 @@ async def build_asset_capability_matrix_with_live_readback(
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 = 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",
)
payload["live_readback_timeout_seconds"] = bounded_timeout_seconds
return payload