From 323e8cc8e0a85eced6d55851765feaef607117c9 Mon Sep 17 00:00:00 2001 From: AWOOOI CD Date: Wed, 15 Jul 2026 05:06:51 +0800 Subject: [PATCH 1/2] chore(cd): deploy 956b55d [skip ci] --- k8s/awoooi-prod/06-deployment-api.yaml | 4 ++-- k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml | 2 +- k8s/awoooi-prod/08-deployment-worker.yaml | 2 +- k8s/awoooi-prod/kustomization.yaml | 4 ++-- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index ffb570d5f..dd1b790d6 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -160,12 +160,12 @@ spec: - name: AWOOOI_BUILD_COMMIT_SHA # 2026-06-29 Codex: CD rewrites this to the deployed image tag so # production deploy readback does not rely on a stale static snapshot. - value: "e4da6570a645e504cc93647d396f45f167b3172a" + value: "956b55d66c9916a36d3d68311053ce32f0d7094e" - name: AWOOOI_DESIRED_API_IMAGE_TAG # 2026-06-30 Codex: CD rewrites this alongside AWOOOI_BUILD_COMMIT_SHA. # Production readback compares runtime image truth against this # GitOps desired tag instead of doing a slow Gitea raw fetch. - value: "e4da6570a645e504cc93647d396f45f167b3172a" + value: "956b55d66c9916a36d3d68311053ce32f0d7094e" - name: DATABASE_POOL_SIZE # 2026-07-01 Codex: production role `awoooi` currently has a low # connection limit. Keep API pool conservative until DB role diff --git a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml index fc010b5a4..209417ee9 100644 --- a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml +++ b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml @@ -66,7 +66,7 @@ spec: - name: DATABASE_NULL_POOL value: "true" - name: AWOOOI_BUILD_COMMIT_SHA - value: "e4da6570a645e504cc93647d396f45f167b3172a" + value: "956b55d66c9916a36d3d68311053ce32f0d7094e" - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER value: "false" - name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index ffb594524..1341ee393 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -173,7 +173,7 @@ spec: - name: DATABASE_BOOTSTRAP_DDL_ENABLED value: "false" - name: AWOOOI_BUILD_COMMIT_SHA - value: "e4da6570a645e504cc93647d396f45f167b3172a" + value: "956b55d66c9916a36d3d68311053ce32f0d7094e" - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER value: "true" - name: ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER diff --git a/k8s/awoooi-prod/kustomization.yaml b/k8s/awoooi-prod/kustomization.yaml index 86efddcaf..9c044cf89 100644 --- a/k8s/awoooi-prod/kustomization.yaml +++ b/k8s/awoooi-prod/kustomization.yaml @@ -40,9 +40,9 @@ resources: # ⚠️ 重要: name 必須與 deployment YAML 中的 image 完全匹配 (含 tag) # newName + newTag 由 CI 透過 kustomize edit set image 注入 images: -- digest: sha256:5c44c63306a02b9375c006060b25297a76fdee42b3b57770a977d1c2fceac5df +- digest: sha256:9ab7979e8052f1361f9852114c3af1e73527875f14206de3bfe09b5dac143f13 name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api -- digest: sha256:4139a54b724e5696c8dfcb079bbf8b4d84baec8bc1bfab03105eeff9842e8bc1 +- digest: sha256:6ae01c8a8202b84c060cdcaf2cc10626951950b1c79069a0063d4f1e02c29e0a name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web From 228d474534fad5202b7b7be1b5f8e2b60ba45e9a Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 01:23:37 +0800 Subject: [PATCH 2/2] fix(api): index asset capability projection --- .../asset_capability_reconciliation_job.py | 3 +- .../ai_automation_asset_capability_matrix.py | 237 +++++++++++------- ...t_ai_automation_asset_capability_matrix.py | 121 +++++++++ 3 files changed, 276 insertions(+), 85 deletions(-) diff --git a/apps/api/src/jobs/asset_capability_reconciliation_job.py b/apps/api/src/jobs/asset_capability_reconciliation_job.py index 5b2ecbac7..33625980f 100644 --- a/apps/api/src/jobs/asset_capability_reconciliation_job.py +++ b/apps/api/src/jobs/asset_capability_reconciliation_job.py @@ -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( 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 6df7a1934..dda2f292b 100644 --- a/apps/api/src/services/ai_automation_asset_capability_matrix.py +++ b/apps/api/src/services/ai_automation_asset_capability_matrix.py @@ -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 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 cdb6acbbe..542701c8c 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/apps/api/tests/test_ai_automation_asset_capability_matrix.py @@ -102,6 +102,93 @@ def test_declared_catalog_expands_to_193_unique_assets() -> None: assert matrix["operation_boundaries"]["raw_asset_key_exposed"] is False +def test_public_projection_counts_live_only_assets_without_embedding_them() -> None: + now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC) + live_assets = [ + { + "asset_id": index + 1, + "asset_key": f"zxq{index:x}v", + "asset_type": "host", + "name": f"zxq{index:x}v", + "last_seen_at": now, + "coverage": {}, + "compliance": {}, + "change_types": [], + } + for index in range(1_000) + ] + + public_matrix = build_asset_capability_matrix( + _single_host_scope(), + live_assets=live_assets, + repo_root=_REPO_ROOT, + generated_at=now, + include_live_only_assets=False, + ) + reconciliation_matrix = build_asset_capability_matrix( + _single_host_scope(), + live_assets=live_assets, + repo_root=_REPO_ROOT, + generated_at=now, + include_live_only_assets=True, + ) + + assert len(public_matrix["assets"]) == 1 + public_summary = public_matrix["summary"] + assert public_summary["live_inventory_asset_count"] == 1_000 + assert public_summary["matrix_asset_count"] == 1 + assert public_summary["reconciled_asset_count"] == 1_001 + assert public_summary["live_only_assets_embedded"] is False + assert public_summary["embedded_live_only_asset_count"] == 0 + assert public_summary["discovered_not_declared_count"] == 1_000 + assert len(reconciliation_matrix["assets"]) == 1_001 + assert reconciliation_matrix["summary"]["live_only_assets_embedded"] is True + assert reconciliation_matrix["summary"]["embedded_live_only_asset_count"] == 1_000 + assert ( + public_matrix["matrix_fingerprint"] + == (reconciliation_matrix["matrix_fingerprint"]) + ) + + +def test_indexed_matching_preserves_score_cap_and_original_live_order() -> None: + now = datetime(2026, 7, 15, 1, 0, tzinfo=UTC) + scope = [ + { + "id": "services", + "label": "Services", + "asset_count": 1, + "asset_ids": ["one two three four five six seven"], + "inventory_sources": ["service-inventory.json"], + } + ] + live_assets = [ + { + "asset_id": 1, + "asset_key": "candidate:first", + "asset_type": "container", + "name": "one two three four five six alpha", + "last_seen_at": now, + }, + { + "asset_id": 2, + "asset_key": "candidate:second", + "asset_type": "container", + "name": "one two three four five six seven beta", + "last_seen_at": now, + }, + ] + + matrix = build_asset_capability_matrix( + scope, + live_assets=live_assets, + repo_root=_REPO_ROOT, + generated_at=now, + include_live_only_assets=False, + ) + + assert matrix["assets"][0]["runtime_identity"]["asset_inventory_id"] == 1 + + def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> None: now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC) initial = build_asset_capability_matrix( @@ -318,6 +405,8 @@ class _FakeLiveReadbackDb: sql = str(statement) bound = params or {} self.calls.append((sql, bound)) + if "set_config('statement_timeout'" in sql: + return _FakeResult([]) if "FROM asset_discovery_run" in sql: return _FakeResult( [ @@ -382,6 +471,13 @@ def test_live_change_event_readback_is_scoped_to_selected_asset_ids( _latest_run, assets, _receipt = asyncio.run(_load_live_rows("awoooi")) + timeout_sql, timeout_params = next( + (sql, params) + for sql, params in db.calls + if "set_config('statement_timeout'" in sql + ) + assert "statement_timeout" in timeout_sql + assert timeout_params == {"statement_timeout": "6000ms"} change_sql, change_params = next( (sql, params) for sql, params in db.calls if "FROM asset_change_event" in sql ) @@ -413,3 +509,28 @@ def test_live_readback_timeout_fails_closed_without_hanging( assert payload["live_readback_timeout_seconds"] == 0.01 assert payload["summary"]["declared_asset_count"] == 193 assert payload["operation_boundaries"]["host_write_performed"] is False + + +def test_matrix_build_phase_is_inside_the_total_timeout(monkeypatch: Any) -> None: + async def _fast_live_readback( + _project_id: str, + ) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]: + return {}, [], {} + + async def _slow_to_thread(*_args: Any, **_kwargs: Any) -> dict[str, Any]: + await asyncio.sleep(1) + return {} + + monkeypatch.setattr(matrix_service, "_load_live_rows", _fast_live_readback) + monkeypatch.setattr(matrix_service.asyncio, "to_thread", _slow_to_thread) + + payload = asyncio.run( + build_asset_capability_matrix_with_live_readback( + repo_root=_REPO_ROOT, + timeout_seconds=0.01, + ) + ) + + assert payload["live_readback_status"] == "degraded" + assert payload["live_readback_error_type"] == "TimeoutError" + assert payload["live_readback_timeout_seconds"] == 0.01