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

fix(automation): close asset reconciliation controls
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 2s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 3m8s
Details
CD Pipeline / build-and-deploy (push) Successful in 14m55s
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-15 09:11:50 +08:00
parent 68602a7489
commit e01db7391e
3 changed files with 357 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -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'