diff --git a/apps/api/src/jobs/asset_capability_reconciliation_job.py b/apps/api/src/jobs/asset_capability_reconciliation_job.py index c13545927..446da976e 100644 --- a/apps/api/src/jobs/asset_capability_reconciliation_job.py +++ b/apps/api/src/jobs/asset_capability_reconciliation_job.py @@ -1,4 +1,4 @@ -"""Daily AIA-P0-006 asset capability reconciliation. +"""AIA-P0-006 asset capability reconciliation. The job projects added, removed, drifted, missing, and stale asset capability states into idempotent automation_operation_log work-item receipts. It writes no @@ -13,9 +13,7 @@ import time import uuid from collections import Counter from collections.abc import Mapping, Sequence -from datetime import datetime, timedelta from typing import Any -from zoneinfo import ZoneInfo import structlog @@ -31,18 +29,17 @@ from src.services.ai_automation_asset_capability_matrix import ( logger = structlog.get_logger(__name__) -_FIRST_DELAY_SECONDS = 90 -_LOOP_BACKOFF_SECONDS = 30 * 60 -_MAX_RECONCILIATION_CANDIDATES = 20_000 +_FIRST_DELAY_SECONDS = 30 +_LOOP_BACKOFF_SECONDS = 5 * 60 +_RECONCILIATION_POLL_SECONDS = 5 * 60 _RECONCILIATION_LIVE_READBACK_TIMEOUT_SECONDS = 30.0 +_WORK_ITEM_INSERT_BATCH_SIZE = 250 +_MAX_RECONCILIATION_CANDIDATES = 20_000 _RETRYABLE_RESULT_STATUSES = { "blocked_safe_no_write", "degraded_cache_refresh", "degraded_no_write", } -_DAILY_TRIGGER_HOUR_TAIPEI = 3 -_DAILY_TRIGGER_MINUTE_TAIPEI = 30 -_TAIPEI = ZoneInfo("Asia/Taipei") def _dict(value: Any) -> dict[str, Any]: @@ -53,19 +50,6 @@ def _canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) -def _seconds_until_next_trigger() -> int: - now = datetime.now(_TAIPEI) - target = now.replace( - hour=_DAILY_TRIGGER_HOUR_TAIPEI, - minute=_DAILY_TRIGGER_MINUTE_TAIPEI, - second=0, - microsecond=0, - ) - if target <= now: - target += timedelta(days=1) - return max(60, int((target - now).total_seconds())) - - def build_reconciliation_summary_payload( matrix: Mapping[str, Any], candidates: Sequence[Mapping[str, Any]], @@ -74,6 +58,7 @@ def build_reconciliation_summary_payload( automation_run_id: str, work_item_id: str, repository_readback_count: int, + work_item_verifier_readback_count: int, created_count: int, existing_count: int, ) -> dict[str, Any]: @@ -89,7 +74,10 @@ def build_reconciliation_summary_payload( "full_scope_class_coverage", ) candidate_count = len(candidates) - readback_verified = repository_readback_count == candidate_count + readback_verified = bool( + repository_readback_count == candidate_count + and work_item_verifier_readback_count == candidate_count + ) mechanism_closed = bool( all(controls.get(control_id) is True for control_id in structural_control_ids) and str(_dict(matrix.get("latest_discovery_run")).get("run_id") or "") @@ -110,6 +98,7 @@ def build_reconciliation_summary_payload( "created_work_item_count": created_count, "existing_work_item_count": existing_count, "repository_readback_count": repository_readback_count, + "work_item_verifier_readback_count": work_item_verifier_readback_count, "repository_readback_verified": readback_verified, "structural_controls": { control_id: controls.get(control_id) is True @@ -135,6 +124,7 @@ def build_reconciliation_summary_payload( "bounded_execution_receipt": { "target": "automation_operation_log", "operation": "idempotent_work_item_projection", + "batch_size": _WORK_ITEM_INSERT_BATCH_SIZE, "created_count": created_count, "existing_count": existing_count, }, @@ -142,6 +132,7 @@ def build_reconciliation_summary_payload( "verifier": "automation_operation_log_row_readback", "expected_count": candidate_count, "observed_count": repository_readback_count, + "verified_work_item_count": work_item_verifier_readback_count, "passed": readback_verified, }, "rollback_or_no_write_terminal": "no_external_runtime_write", @@ -186,12 +177,11 @@ def build_reconciliation_work_item_input( async def run_asset_capability_reconciliation_loop() -> None: - """Run shortly after startup, then at 03:30 Asia/Taipei every day.""" + """Reconcile shortly after startup and retry each discovery fingerprint.""" logger.info( "asset_capability_reconciliation_loop_started", - daily_trigger_hour_taipei=_DAILY_TRIGGER_HOUR_TAIPEI, - daily_trigger_minute_taipei=_DAILY_TRIGGER_MINUTE_TAIPEI, + poll_interval_seconds=_RECONCILIATION_POLL_SECONDS, ) await asyncio.sleep(_FIRST_DELAY_SECONDS) triggered_by = "startup" @@ -216,8 +206,8 @@ async def run_asset_capability_reconciliation_loop() -> None: await asyncio.sleep(_LOOP_BACKOFF_SECONDS) triggered_by = "retry" continue - await asyncio.sleep(_seconds_until_next_trigger()) - triggered_by = "cron" + await asyncio.sleep(_RECONCILIATION_POLL_SECONDS) + triggered_by = "poll" async def reconcile_once( @@ -227,13 +217,6 @@ async def reconcile_once( ) -> dict[str, Any]: """Reconcile the matrix and persist idempotent gap/drift work items.""" - from src.services.ai_advisory_helpers import try_acquire_daily_lock - - if triggered_by == "cron" and not await try_acquire_daily_lock( - "asset_capability_reconciliation" - ): - return {"status": "skipped", "reason": "not_leader"} - started = time.monotonic() matrix = await build_asset_capability_matrix_with_live_readback( project_id=project_id, @@ -247,6 +230,7 @@ async def reconcile_once( ) return { "status": "degraded_no_write", + "triggered_by": triggered_by, "live_readback_status": matrix.get("live_readback_status"), "candidate_count": 0, } @@ -310,17 +294,43 @@ async def _persist_reconciliation( from src.db.base import get_db_context matrix_fingerprint = str(matrix.get("matrix_fingerprint") or "") - taipei_date = datetime.now(_TAIPEI).date().isoformat() + discovery_run_id = str( + _dict(matrix.get("latest_discovery_run")).get("run_id") or "" + ) + candidate_rows = [dict(candidate) for candidate in candidates] + expected_fingerprint_list = [ + str(candidate.get("candidate_fingerprint") or "") + for candidate in candidate_rows + ] + expected_fingerprints = set(expected_fingerprint_list) + if not matrix_fingerprint: + raise ValueError("reconciliation_matrix_fingerprint_missing") + if not discovery_run_id: + raise ValueError("reconciliation_discovery_run_missing") + if ( + any(not fingerprint for fingerprint in expected_fingerprint_list) + or len(expected_fingerprints) != len(candidate_rows) + ): + raise ValueError("reconciliation_candidate_fingerprint_invalid") + idempotency_key = ( - f"asset-capability-reconciliation:{taipei_date}:{matrix_fingerprint}" + "asset-capability-reconciliation:" + f"{discovery_run_id}:{matrix_fingerprint}" ) trace_id = f"asset-capability:{matrix_fingerprint[:16]}" automation_run_id = str(uuid.uuid4()) work_item_id = "AIA-P0-006" - discovery_run_id = str( - _dict(matrix.get("latest_discovery_run")).get("run_id") or "" + fingerprint_readback_sql = text( + """ + SELECT DISTINCT input ->> 'candidate_fingerprint' AS candidate_fingerprint + FROM automation_operation_log + WHERE actor = :actor + AND COALESCE(input ->> 'semantic_operation_type', operation_type) + = :semantic_operation_type + AND input ->> 'candidate_fingerprint' + = ANY(CAST(:candidate_fingerprints AS text[])) + """ ) - created_count = 0 async with get_db_context(project_id) as db: await db.execute( @@ -338,6 +348,9 @@ async def _persist_reconciliation( AND input ->> 'idempotency_key' = :idempotency_key AND status = 'success' AND output ->> 'closed' = 'true' + AND output ->> 'repository_readback_verified' = 'true' + AND output ->> 'candidate_count' + = output ->> 'work_item_verifier_readback_count' ORDER BY created_at DESC LIMIT 1 """ @@ -352,38 +365,23 @@ async def _persist_reconciliation( if isinstance(prior, dict): return prior - candidate_fingerprints = [ - str(candidate.get("candidate_fingerprint") or "") - for candidate in candidates - ] - existing_fingerprints: set[str] = set() - if candidate_fingerprints: - existing = await db.execute( - text( - """ - SELECT DISTINCT input ->> 'candidate_fingerprint' - FROM automation_operation_log - WHERE actor = :actor - AND COALESCE(input ->> 'semantic_operation_type', operation_type) - = :semantic_operation_type - AND input ->> 'candidate_fingerprint' - = ANY(CAST(:candidate_fingerprints AS text[])) - """ - ), + persisted_before: set[str] = set() + if expected_fingerprint_list: + existing_result = await db.execute( + fingerprint_readback_sql, { "actor": RECONCILIATION_ACTOR, "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, - "candidate_fingerprints": candidate_fingerprints, + "candidate_fingerprints": expected_fingerprint_list, }, ) - existing_fingerprints = { - str(row[0]) for row in existing.fetchall() if row[0] + persisted_before = { + str(row[0]) for row in existing_result.fetchall() if row[0] } - insert_rows: list[dict[str, Any]] = [] - for candidate_raw in candidates: - candidate = dict(candidate_raw) - if candidate["candidate_fingerprint"] in existing_fingerprints: + insert_parameters: list[dict[str, Any]] = [] + for candidate in candidate_rows: + if candidate["candidate_fingerprint"] in persisted_before: continue input_payload = build_reconciliation_work_item_input( candidate, @@ -398,7 +396,7 @@ async def _persist_reconciliation( "post_verifier": "pending_batch_readback", "rollback_or_no_write_terminal": "no_external_runtime_write", } - insert_rows.append( + insert_parameters.append( { "asset_id": candidate.get("asset_inventory_id"), "discovery_run_id": discovery_run_id, @@ -409,7 +407,9 @@ async def _persist_reconciliation( { "would_create_work_item": True, "external_runtime_write": False, - "candidate_fingerprint": candidate["candidate_fingerprint"], + "candidate_fingerprint": candidate[ + "candidate_fingerprint" + ], } ), "tags": [ @@ -420,74 +420,134 @@ async def _persist_reconciliation( } ) - if insert_rows: - await db.execute( - text( - """ - INSERT INTO automation_operation_log ( - operation_type, - asset_id, - run_id, - actor, - input, - output, - dry_run_result, - status, - tags - ) VALUES ( - 'asset_discovered', - :asset_id, - CAST(NULLIF(:discovery_run_id, '') AS uuid), - :actor, - CAST(:input AS jsonb), - CAST(:output AS jsonb), - CAST(:dry_run_result AS jsonb), - 'success', - :tags - ) - """ - ), - insert_rows, + insert_work_item_sql = text( + """ + INSERT INTO automation_operation_log ( + operation_type, + asset_id, + run_id, + actor, + input, + output, + dry_run_result, + status, + tags + ) VALUES ( + 'asset_discovered', + :asset_id, + CAST(NULLIF(:discovery_run_id, '') AS uuid), + :actor, + CAST(:input AS jsonb), + CAST(:output AS jsonb), + CAST(:dry_run_result AS jsonb), + 'success', + :tags + ) + """ + ) + for batch_start in range( + 0, + len(insert_parameters), + _WORK_ITEM_INSERT_BATCH_SIZE, + ): + await db.execute( + insert_work_item_sql, + insert_parameters[ + batch_start : batch_start + _WORK_ITEM_INSERT_BATCH_SIZE + ], ) - created_count = len(insert_rows) - expected_fingerprints = { - str(candidate.get("candidate_fingerprint") or "") - for candidate in candidates - } persisted_fingerprints: set[str] = set() - if candidate_fingerprints: + if expected_fingerprint_list: rows_result = await db.execute( - text( - """ - SELECT DISTINCT input ->> 'candidate_fingerprint' - FROM automation_operation_log - WHERE actor = :actor - AND COALESCE(input ->> 'semantic_operation_type', operation_type) - = :semantic_operation_type - AND input ->> 'candidate_fingerprint' - = ANY(CAST(:candidate_fingerprints AS text[])) - """ - ), + fingerprint_readback_sql, { "actor": RECONCILIATION_ACTOR, "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, - "candidate_fingerprints": candidate_fingerprints, + "candidate_fingerprints": expected_fingerprint_list, }, ) persisted_fingerprints = { str(row[0]) for row in rows_result.fetchall() if row[0] } repository_readback_count = len(expected_fingerprints & persisted_fingerprints) + work_item_verifier_readback_count = 0 + if repository_readback_count == len(expected_fingerprints): + if expected_fingerprint_list: + await db.execute( + text( + """ + UPDATE automation_operation_log + SET output = COALESCE(output, '{}'::jsonb) + || CAST(:verification_output AS jsonb) + WHERE actor = :actor + AND COALESCE( + input ->> 'semantic_operation_type', operation_type + ) = :semantic_operation_type + AND input ->> 'candidate_fingerprint' + = ANY(CAST(:candidate_fingerprints AS text[])) + AND COALESCE( + output ->> 'repository_readback_verified', 'false' + ) <> 'true' + """ + ), + { + "actor": RECONCILIATION_ACTOR, + "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, + "candidate_fingerprints": expected_fingerprint_list, + "verification_output": _canonical_json( + { + "post_verifier": "passed_repository_readback", + "repository_readback_verified": True, + "verified_by": ( + "asset_capability_reconciler_batch_readback" + ), + "verified_run_id": automation_run_id, + } + ), + }, + ) + verified_result = await db.execute( + text( + """ + SELECT DISTINCT + input ->> 'candidate_fingerprint' AS candidate_fingerprint + FROM automation_operation_log + WHERE actor = :actor + AND COALESCE( + input ->> 'semantic_operation_type', operation_type + ) = :semantic_operation_type + AND input ->> 'candidate_fingerprint' + = ANY(CAST(:candidate_fingerprints AS text[])) + AND output ->> 'repository_readback_verified' = 'true' + """ + ), + { + "actor": RECONCILIATION_ACTOR, + "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, + "candidate_fingerprints": expected_fingerprint_list, + }, + ) + verified_fingerprints = { + str(row[0]) for row in verified_result.fetchall() if row[0] + } + work_item_verifier_readback_count = len( + expected_fingerprints & verified_fingerprints + ) + + created_count = len(insert_parameters) summary_output = build_reconciliation_summary_payload( matrix, - candidates, + candidate_rows, trace_id=trace_id, automation_run_id=automation_run_id, work_item_id=work_item_id, repository_readback_count=repository_readback_count, + work_item_verifier_readback_count=( + work_item_verifier_readback_count + ), created_count=created_count, - existing_count=len(candidates) - created_count, + existing_count=len(candidate_rows) - created_count, ) summary_input = { "schema_version": RECONCILIATION_SCHEMA_VERSION, 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 30ab3ac40..6bd56a430 100644 --- a/apps/api/src/services/ai_automation_asset_capability_matrix.py +++ b/apps/api/src/services/ai_automation_asset_capability_matrix.py @@ -627,6 +627,9 @@ 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 + ) created_at = row.get("created_at") fresh = bool( row @@ -647,6 +650,7 @@ def _receipt_controls( fresh and output.get("repository_readback_verified") is True and candidate_count == readback_count + and candidate_count == verifier_readback_count ), } @@ -912,6 +916,9 @@ def build_asset_capability_matrix( "reconciliation_work_item_readback_count": int( receipt_output.get("repository_readback_count") or 0 ), + "reconciliation_work_item_verifier_readback_count": int( + receipt_output.get("work_item_verifier_readback_count") or 0 + ), }, "scope_rollups": [ { @@ -945,6 +952,9 @@ def build_asset_capability_matrix( "repository_readback_verified" ) is True, + "work_item_verifier_readback_count": int( + receipt_output.get("work_item_verifier_readback_count") or 0 + ), }, "operation_boundaries": { "read_only_api": True, @@ -1074,7 +1084,7 @@ async def _load_live_rows( cs.coverage_status, cs.created_at AS coverage_created_at FROM asset_inventory ai - LEFT JOIN asset_coverage_snapshot cs + 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' diff --git a/apps/api/src/services/awoooi_priority_work_order_readback.py b/apps/api/src/services/awoooi_priority_work_order_readback.py index 3b0232322..e85e5fded 100644 --- a/apps/api/src/services/awoooi_priority_work_order_readback.py +++ b/apps/api/src/services/awoooi_priority_work_order_readback.py @@ -798,7 +798,7 @@ _AI_AUTOMATION_PROGRAM_WORK_ITEMS: list[dict[str, Any]] = [ "problem": "Gitea runner 仍會把 actions/checkout 等 action 從 github.com 抓入執行環境，GitHub freeze 尚未覆蓋 CI action resolution。", "professional_practice": "self-hosted action registry、immutable digest pinning、SBOM/provenance、network egress verification", "asset_scope_ids": ["products", "tools", "packages", "observability_and_security"], - "observed_awoooi_gitea_workflow_file_count": 13, + "observed_awoooi_gitea_workflow_file_count": 14, "observed_awoooi_external_action_reference_count": 18, "acceptance": [ "所有產品的 Gitea workflow action 只由受控 Gitea mirror 或 repo-owned immutable action 供應", 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 ac817d4c5..0e9d3185f 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/apps/api/tests/test_ai_automation_asset_capability_matrix.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import inspect +import json from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any @@ -214,6 +215,7 @@ def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> N "matrix_fingerprint": initial["matrix_fingerprint"], "candidate_count": 0, "repository_readback_count": 0, + "work_item_verifier_readback_count": 0, "repository_readback_verified": True, "closed": True, }, @@ -334,6 +336,7 @@ def test_reconciliation_summary_requires_repository_readback() -> None: automation_run_id="run-1", work_item_id="AIA-P0-006", repository_readback_count=0, + work_item_verifier_readback_count=0, created_count=1, existing_count=0, ) @@ -344,6 +347,7 @@ def test_reconciliation_summary_requires_repository_readback() -> None: automation_run_id="run-1", work_item_id="AIA-P0-006", repository_readback_count=1, + work_item_verifier_readback_count=1, created_count=1, existing_count=0, ) @@ -354,6 +358,31 @@ def test_reconciliation_summary_requires_repository_readback() -> None: assert passed["closed"] is True +def test_reconciliation_summary_requires_each_work_item_verifier() -> None: + matrix = build_asset_capability_matrix( + _single_host_scope(), + latest_run={"run_id": "discovery-run-1", "status": "success"}, + repo_root=_REPO_ROOT, + ) + candidates = build_reconciliation_candidates(matrix) + + payload = build_reconciliation_summary_payload( + matrix, + candidates, + trace_id="trace-1", + automation_run_id="run-1", + work_item_id="AIA-P0-006", + repository_readback_count=1, + work_item_verifier_readback_count=0, + created_count=1, + existing_count=0, + ) + + assert payload["repository_readback_verified"] is False + assert payload["post_verifier"]["passed"] is False + assert payload["closed"] is False + + def test_reconciliation_work_item_uses_same_trace_and_run_as_summary() -> None: matrix = build_asset_capability_matrix( _single_host_scope(), @@ -380,13 +409,15 @@ def test_reconciliation_work_item_uses_same_trace_and_run_as_summary() -> None: def test_reconciliation_persistence_batches_candidate_queries_and_inserts() -> None: source = inspect.getsource(_persist_reconciliation) - loop_start = source.index("for candidate_raw in candidates:") - loop_end = source.index("if insert_rows:") + loop_start = source.index("for candidate in candidate_rows:") + loop_end = source.index("insert_work_item_sql =") assert "ANY(CAST(:candidate_fingerprints AS text[]))" in source - assert "insert_rows" in source + assert "insert_parameters" in source + assert "_WORK_ITEM_INSERT_BATCH_SIZE" in source assert "await db.execute" not in source[loop_start:loop_end] assert "AND output ->> 'closed' = 'true'" in source + assert "work_item_verifier_readback_count" in source def test_reconcile_once_uses_bounded_worker_live_readback_timeout( @@ -474,7 +505,7 @@ def test_reconcile_once_publishes_closed_public_matrix_cache( assert result["public_cache_refreshed"] is True -def test_reconciliation_loop_retries_degraded_result_before_daily_wait( +def test_reconciliation_loop_retries_degraded_result_before_poll( monkeypatch: Any, ) -> None: triggered_by_values: list[str] = [] @@ -495,7 +526,7 @@ def test_reconciliation_loop_retries_degraded_result_before_daily_wait( monkeypatch.setattr(reconciliation_job.asyncio, "sleep", _fake_sleep) monkeypatch.setattr(reconciliation_job, "_FIRST_DELAY_SECONDS", 90) monkeypatch.setattr(reconciliation_job, "_LOOP_BACKOFF_SECONDS", 1_800) - monkeypatch.setattr(reconciliation_job, "_seconds_until_next_trigger", lambda: 1) + monkeypatch.setattr(reconciliation_job, "_RECONCILIATION_POLL_SECONDS", 1) try: asyncio.run(reconciliation_job.run_asset_capability_reconciliation_loop()) @@ -624,6 +655,136 @@ class _FakeDbContext: return None +class _FakeReconciliationResult: + def __init__( + self, + *, + scalar: Any = None, + rows: list[tuple[Any, ...]] | None = None, + ) -> None: + self._scalar = scalar + self._rows = rows or [] + + def scalar_one_or_none(self) -> Any: + return self._scalar + + def scalar_one(self) -> Any: + return self._scalar + + def fetchall(self) -> list[tuple[Any, ...]]: + return self._rows + + +class _FakeReconciliationDb: + def __init__(self) -> None: + self.persisted_fingerprints: set[str] = set() + self.verified_fingerprints: set[str] = set() + self.summary_outputs: list[dict[str, Any]] = [] + self.work_item_batch_sizes: list[int] = [] + + async def execute( + self, + statement: Any, + params: dict[str, Any] | list[dict[str, Any]] | None = None, + ) -> _FakeReconciliationResult: + sql = str(statement) + if "pg_advisory_xact_lock" in sql: + return _FakeReconciliationResult() + if "SELECT output" in sql: + prior = self.summary_outputs[-1] if self.summary_outputs else None + return _FakeReconciliationResult(scalar=prior) + if "INSERT INTO automation_operation_log" in sql and isinstance(params, list): + self.work_item_batch_sizes.append(len(params)) + for row in params: + payload = json.loads(str(row["input"])) + self.persisted_fingerprints.add(payload["candidate_fingerprint"]) + return _FakeReconciliationResult() + if "UPDATE automation_operation_log" in sql: + assert isinstance(params, dict) + requested = set(params["candidate_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) + requested = set(params["candidate_fingerprints"]) + source = ( + self.verified_fingerprints + if "repository_readback_verified" in sql + else self.persisted_fingerprints + ) + return _FakeReconciliationResult( + rows=[(fingerprint,) for fingerprint in sorted(requested & source)] + ) + if "RETURNING op_id::text" in sql: + assert isinstance(params, dict) + self.summary_outputs.append( + json.loads(str(params["output"])) + ) + return _FakeReconciliationResult(scalar="summary-op-1") + raise AssertionError(f"unexpected SQL: {sql}") + + +class _FakeReconciliationDbContext: + def __init__(self, db: _FakeReconciliationDb) -> None: + self.db = db + + async def __aenter__(self) -> _FakeReconciliationDb: + return self.db + + async def __aexit__(self, *_args: object) -> None: + return None + + +def test_reconciliation_persists_bounded_batches_and_reuses_verified_summary( + monkeypatch: Any, +) -> None: + db = _FakeReconciliationDb() + monkeypatch.setattr( + "src.db.base.get_db_context", + lambda _project_id: _FakeReconciliationDbContext(db), + ) + 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] + }, + } + candidates = [ + { + "candidate_fingerprint": f"fingerprint-{index:04d}", + "candidate_kind": "capability_gap", + "canonical_asset_id": f"awoooi:services:asset-{index:04d}", + "asset_inventory_id": index, + "missing_stage_ids": ["sensor"], + "stale_stage_ids": [], + "change_types": [], + } + for index in range(301) + ] + + first = asyncio.run( + _persist_reconciliation(matrix, candidates, project_id="awoooi") + ) + second = asyncio.run( + _persist_reconciliation(matrix, candidates, project_id="awoooi") + ) + + assert db.work_item_batch_sizes == [250, 51] + assert first["candidate_count"] == 301 + assert first["repository_readback_count"] == 301 + assert first["work_item_verifier_readback_count"] == 301 + assert first["repository_readback_verified"] is True + assert first["closed"] is True + assert second["repository_readback_verified"] is True + assert len(db.summary_outputs) == 1 + + def test_live_change_event_readback_is_scoped_to_selected_asset_ids( monkeypatch: Any, ) -> None: @@ -647,6 +808,11 @@ def test_live_change_event_readback_is_scoped_to_selected_asset_ids( change_sql, change_params = next( (sql, params) for sql, params in db.calls if "FROM asset_change_event" in sql ) + inventory_sql = next( + sql for sql, _params in db.calls if "FROM asset_inventory" in sql + ) + assert "JOIN asset_coverage_snapshot" in inventory_sql + assert "LEFT JOIN asset_coverage_snapshot" not in inventory_sql assert "asset_id = ANY(:asset_ids)" in change_sql assert change_params == {"asset_ids": [42]} assert assets[0]["change_types"] == ["asset_modified"] diff --git a/docs/LOGBOOK.md b/docs/LOGBOOK.md index 8f4eb97c4..bd7166202 100644 --- a/docs/LOGBOOK.md +++ b/docs/LOGBOOK.md @@ -1,3 +1,25 @@ +## 2026-07-15 — P0 MCP Playwright immutable mirror 與獨立供應鏈 verifier + +**完成內容**： +- mcp.so 僅保留 discovery URL；可執行 bytes 只從 policy allowlist 的 npm registry exact URLs 取得。`@playwright/mcp=0.0.78`、`playwright` / `playwright-core=1.62.0-alpha-1783623505000` 全部固定 version、SRI SHA-512、SHA-1、npm ECDSA signature、SLSA subject、dependency closure 與 Apache-2.0 license。 +- 新增 fail-closed artifact controller：拒絕 redirect、非 allowlist host、floating version、dependency/metadata drift、unsafe/duplicate tar entry、已知 OSV vulnerability 與未通過簽章的 artifact；產生 deterministic CycloneDX 1.5 SBOM 與 scratch OCI data bundle，不執行 package install、MCP、browser 或 container。 +- 新增完全不 import controller 的獨立 verifier：由 Harbor immutable digest 重新 pull/save image、只把 layer 當資料讀取，限制 file/byte/path/symlink/whiteout，要求 exact 8-file bundle，再次重算 manifest/SBOM/tarball hash、以 OpenSSL 重驗 npm signature 並重解 SLSA DSSE subject。 +- Gitea `mcp-external-artifact-mirror.yaml` 不使用遠端 action；push、manual 與每週三 pinned revalidation 都先跑 controller/verifier tests、host-pressure gate、Harbor password-stdin、獨立 readback與 logout。receipt 以 ephemeral askpass normal push 到專用 `mcp-artifact-receipts` audit branch，不再推 main、取消有效 CD 或保存 credential。 +- Gitea run `#5174` 成功；controlled run `dce032df-0a9a-476f-940b-e61a94865e1b` 產生 internal digest `sha256:de002c418bbb94fb4609539c4259c39ccb7f8b353f8bac8fbcba9e5742ad79ed`，獨立 verifier 為 `completed_internal_mirror_verified`，audit branch commit `9c08ada5c0f15bb6924304af29cd13c8376182ff`。 + +**source/test evidence**： +- artifact controller/verifier focused tests `24 passed`；MCP control-plane/federation/version-lifecycle `20 passed`。 +- 新增第 14 條 Gitea workflow 後，CD `#5173` 以 `2627 passed / 1 failed` 找出 workflow health contract 仍固定 13；已同步 workflow health snapshot、secret-name inventory、Gitea capability readback 與 priority work-order count，聚焦回歸 `40 passed`，secret names 維持 `26`、secret values collected=`false`。 +- Ruff、py_compile、JSON/YAML parse、freeze/floating-reference guard 與 `git diff --check` 通過。 + +**尚未完成 / 不誤報**： +- 目前只完成 immutable internal mirror 與供應鏈 verifier；`deployment_allowed=false`、`replay_allowed=false`、`shadow_allowed=false`、`canary_allowed=false`。尚未實作 registered public-origin replay adapter、compatibility replay、shadow/canary、runtime upgrade、post-verifier 或 rollback execution。 +- 外部 catalog / tarball / attestation 未寫 RAG；只保留 normalized digest、SBOM 與 supply-chain receipt。未讀 secret value、`.env`、raw session、SQLite/auth；未連線或下載 GitHub / Actions / hosted artifact，也未使用 `npx -y`、`@latest`、`:latest`、force push 或 production route mutation。 +- 本筆 inventory drift 修正與 catalog mirror state 尚待最新 main CD/deploy marker/production UI readback；不得由 `#5174` mirror success 倒推 AWOOOI production 已部署。 + +**下一步**： +- 先取得包含本修正的 Gitea CD terminal與 production MCP page readback；再以 public-only origin allowlist 建立 repo-owned replay adapter，要求 compatibility、prompt-injection、latency、accessibility、session/file/download no-write verifier與 rollback receipt 全數通過，才可進第一個 noncritical shadow/canary。 + ## 2026-07-15 — P0 MCP EwoooC/MOMO durable production federation source closure **完成內容**： diff --git a/docs/operations/mcp-control-plane-catalog.snapshot.json b/docs/operations/mcp-control-plane-catalog.snapshot.json index 312253008..7b5d6576b 100644 --- a/docs/operations/mcp-control-plane-catalog.snapshot.json +++ b/docs/operations/mcp-control-plane-catalog.snapshot.json @@ -301,18 +301,21 @@ "data_boundary": "public_and_test_surfaces", "deployment_allowed": false, "version_state": "exact_0.0.78_check_verified", - "digest_state": "upstream_sha512_pinned_internal_mirror_pending", - "sbom_state": "cyclonedx_1.5_check_verified_internal_mirror_pending", - "signature_state": "npm_registry_ecdsa_and_slsa_subject_verified", + "digest_state": "internal_mirror_digest_verified_sha256_de002c418bbb94fb4609539c4259c39ccb7f8b353f8bac8fbcba9e5742ad79ed", + "sbom_state": "cyclonedx_1.5_mirror_readback_verified", + "signature_state": "npm_registry_ecdsa_and_slsa_subject_independently_reverified", "rag_policy": "screenshots_and_accessibility_receipts_only", "blockers": [ - "internal_mirror_and_independent_digest_readback_pending", "registered_public_origin_replay_adapter_missing", + "compatibility_replay_not_executed", "shadow_canary_verifier_and_rollback_execution_pending" ], "evidence_refs": [ "config/mcp/playwright-mcp-artifact-policy.json", - "docs/operations/external-mcp-artifacts/playwright-mcp-0.0.78-check.snapshot.json" + "docs/operations/external-mcp-artifacts/playwright-mcp-0.0.78-check.snapshot.json", + "gitea:mcp-artifact-receipts@9c08ada5c0f15bb6924304af29cd13c8376182ff:docs/operations/external-mcp-artifacts/playwright-mcp-0.0.78-controller.snapshot.json", + "gitea:mcp-artifact-receipts@9c08ada5c0f15bb6924304af29cd13c8376182ff:docs/operations/external-mcp-artifacts/playwright-mcp-0.0.78-verifier.snapshot.json", + "registry.wooo.work/awoooi/mcp-artifacts/playwright-mcp@sha256:de002c418bbb94fb4609539c4259c39ccb7f8b353f8bac8fbcba9e5742ad79ed" ] }, { diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index 3af3f9487..6dca50685 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: "0f2ddc0ce81ab52c7fbe33cb2e14dc54d0d33ed5" + value: "e01db7391e89a52958d9c3f06c3218fdc5053eb7" - 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: "0f2ddc0ce81ab52c7fbe33cb2e14dc54d0d33ed5" + value: "e01db7391e89a52958d9c3f06c3218fdc5053eb7" - 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 643fc8be7..683b639a9 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: "0f2ddc0ce81ab52c7fbe33cb2e14dc54d0d33ed5" + value: "e01db7391e89a52958d9c3f06c3218fdc5053eb7" - 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 edad2bbb6..c9a589790 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -177,7 +177,7 @@ spec: - name: DATABASE_BOOTSTRAP_DDL_ENABLED value: "false" - name: AWOOOI_BUILD_COMMIT_SHA - value: "0f2ddc0ce81ab52c7fbe33cb2e14dc54d0d33ed5" + value: "e01db7391e89a52958d9c3f06c3218fdc5053eb7" - 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 2d19b025a..dbf720196 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:1d195016008e4f5ec795170055aca2194161f74fc44fb373d387f26417f0b4af +- digest: sha256:a06af69531087ef123b91bcae5f4170f322035156a64e5d09ef8ebdc23ea1c7a name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api -- digest: sha256:a1422caaefe56ce45d185cb85147ab3591e25251dc421cf738f1439565348c4d +- digest: sha256:9e6ada8cef17243c83230d7890acc1da154add49569b407a590e633d5a5ea70a name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web