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

fix(agent): keep alert replay closure monotonic
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 1m11s
Details
CD Pipeline / build-and-deploy (push) Successful in 7m28s
Details
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 0s
Details
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 8s
Details
CD Pipeline / post-deploy-checks (push) Successful in 2m19s
Details

Browse Source
This commit is contained in:
ogt
2026-07-14 11:24:26 +08:00
parent 82bb69b38f
commit afe15662f3
7 changed files with 690 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

294
apps/api/src/services/ai_agent_autonomous_runtime_control.py
View File

@@ -5135,6 +5135,185 @@ def _build_autonomous_retry_rollback_terminal_readback(
}
def _build_stdin_boundary_replay_runtime_readback(
operation_latest_rows: Iterable[Mapping[str, Any] | Any],
) -> dict[str, Any]:
"""Expose a failed-check replay without promoting it to primary flow."""
rows = [_row_mapping(row) for row in operation_latest_rows]
replay_checks = [
row
for row in rows
if row.get("execution_mode") == "stdin_boundary_check_mode_replay"
]
replay_terminals = [
row
for row in rows
if row.get("execution_mode")
== "stdin_boundary_check_mode_replay_terminal"
]
if not replay_terminals:
return {
"schema_version": "awoooi_stdin_boundary_replay_runtime_v1",
"status": "not_observed_in_bounded_read",
"latest_replay_attempt_closed": False,
"automation_run_id": None,
"retry_check_mode_op_id": None,
"retry_terminal_op_id": None,
"terminal_type": None,
"check_mode_replay_returncode": None,
"next_ai_action": "continue_bounded_no_write_historical_replay",
"controls": {},
"active_blockers": ["stdin_boundary_replay_terminal_not_observed"],
"bounded_read_counts": {
"replay_check_count": len(replay_checks),
"no_write_terminal_count": 0,
},
"operation_boundaries": {
"writes_on_read": False,
"runtime_apply_executed_by_replay": False,
"raw_output_returned": False,
"secret_value_read": False,
"primary_runtime_flow_replaced": False,
},
}
terminal = replay_terminals[0]
terminal_op_id = str(terminal.get("op_id") or "")
retry_check_mode_op_id = str(terminal.get("parent_op_id") or "")
replay_check = _operation_by_id(rows, retry_check_mode_op_id) or {}
automation_run_id = str(terminal.get("automation_run_id") or "")
raw_receipts = terminal.get("runtime_stage_receipts")
if isinstance(raw_receipts, str):
try:
raw_receipts = json.loads(raw_receipts)
except json.JSONDecodeError:
raw_receipts = []
if not isinstance(raw_receipts, list):
raw_receipts = []
retry_receipt = next(
(
dict(receipt)
for receipt in raw_receipts
if isinstance(receipt, Mapping)
and receipt.get("stage_id") == "retry_or_rollback"
),
{},
)
detail = retry_receipt.get("detail")
if not isinstance(detail, Mapping):
detail = {}
failed_check_mode_op_id = str(detail.get("failed_check_mode_op_id") or "")
terminal_type = str(detail.get("terminal_type") or "")
runtime_apply_executed = detail.get("runtime_apply_executed") is True
controls = {
"historical_replay_check_correlated": bool(
replay_check
and _bool_value(
replay_check.get("historical_stdin_boundary_replay")
)
and replay_check.get("execution_mode")
== "stdin_boundary_check_mode_replay"
and str(replay_check.get("replay_of_check_mode_op_id") or "")
== failed_check_mode_op_id
),
"same_run_correlation": bool(
automation_run_id
and str(replay_check.get("automation_run_id") or "")
== automation_run_id
and str(retry_receipt.get("automation_run_id") or "")
== automation_run_id
),
"retry_terminal_receipt_contract": bool(
retry_receipt.get("schema_version")
== AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION
and retry_receipt.get("durable_receipt") is True
and str(retry_receipt.get("evidence_ref") or "")
and detail.get("schema_version")
== "ansible_failed_check_stdin_boundary_retry_terminal_v1"
and str(detail.get("retry_check_mode_op_id") or "")
== retry_check_mode_op_id
and str(detail.get("retry_terminal_op_id") or "")
== terminal_op_id
),
"bounded_single_retry": bool(
detail.get("bounded_retry") is True
and _int_value(detail.get("retry_attempt")) == 1
and _int_value(detail.get("max_retry_attempts")) == 1
),
"check_mode_replay_performed": (
detail.get("check_mode_replay_performed") is True
),
"runtime_apply_not_executed": (
detail.get("runtime_apply_executed") is False
),
"verified_no_write_terminal": (
detail.get("verified_no_write_terminal") is True
),
"repository_readback_verified": bool(
detail.get("retry_operation_readback_verified") is True
and detail.get("repository_readback_verified") is True
and detail.get("durable_write_acknowledged") is True
),
"public_safe_receipt": bool(
detail.get("raw_log_payload_stored") is False
and detail.get("secret_value_stored") is False
),
}
active_blockers = [
f"{control_id}_not_verified"
for control_id, verified in controls.items()
if verified is not True
]
closed = not active_blockers
replay_failed = "failed_waiting_transport_repair" in terminal_type
evidence_refs = [
f"automation-run:{automation_run_id}" if automation_run_id else "",
(
f"retry-check:{retry_check_mode_op_id}"
if retry_check_mode_op_id
else ""
),
f"retry-terminal:{terminal_op_id}" if terminal_op_id else "",
]
return {
"schema_version": "awoooi_stdin_boundary_replay_runtime_v1",
"status": (
"verified_no_write_terminal_repair_required"
if closed and replay_failed
else "verified_no_write_terminal_requeue_ready"
if closed
else "partial_replay_terminal_receipt"
),
"latest_replay_attempt_closed": closed,
"automation_run_id": automation_run_id or None,
"failed_check_mode_op_id": failed_check_mode_op_id or None,
"retry_check_mode_op_id": retry_check_mode_op_id or None,
"retry_terminal_op_id": terminal_op_id or None,
"terminal_type": terminal_type or None,
"check_mode_replay_returncode": detail.get(
"check_mode_replay_returncode"
),
"next_ai_action": detail.get("safe_next_action"),
"controls": controls,
"active_blockers": active_blockers,
"bounded_read_counts": {
"replay_check_count": len(replay_checks),
"no_write_terminal_count": len(replay_terminals),
},
"evidence_refs": [ref for ref in evidence_refs if ref],
"operation_boundaries": {
"writes_on_read": False,
"runtime_apply_executed_by_replay": runtime_apply_executed,
"raw_output_returned": False,
"secret_value_read": False,
"primary_runtime_flow_replaced": False,
},
}
def build_runtime_receipt_readback_from_rows(
*,
project_id: str = _DEFAULT_PROJECT_ID,
@@ -5267,6 +5446,9 @@ def build_runtime_receipt_readback_from_rows(
retry_rollback_terminal = _build_autonomous_retry_rollback_terminal_readback(
retry_terminal_latest
)
stdin_boundary_replay = _build_stdin_boundary_replay_runtime_readback(
operation_latest
)
loop_ledger = _autonomous_execution_loop_ledger(
project_id=project_id,
operation_latest_rows=operation_latest,
@@ -5527,6 +5709,7 @@ def build_runtime_receipt_readback_from_rows(
"latest_failure_classification": latest_failure,
"controlled_retry_package": retry_package,
"autonomous_retry_rollback_terminal": retry_rollback_terminal,
"historical_stdin_boundary_replay": stdin_boundary_replay,
"autonomous_execution_loop_ledger": loop_ledger,
"trace_ledger": trace_ledger,
"log_integration_taxonomy": log_integration_taxonomy,
@@ -5561,6 +5744,14 @@ def _attach_runtime_receipt_readback(
payload["autonomous_retry_rollback_terminal"] = dict(
retry_rollback_terminal
)
stdin_boundary_replay = readback.get(
"historical_stdin_boundary_replay"
)
if not isinstance(stdin_boundary_replay, Mapping):
stdin_boundary_replay = {}
payload["historical_stdin_boundary_replay"] = dict(
stdin_boundary_replay
)
log_executor = readback.get("log_controlled_writeback_executor")
if not isinstance(log_executor, Mapping):
log_executor = {}
@@ -5659,6 +5850,20 @@ def _attach_runtime_receipt_readback(
"autonomous_retry_rollback_terminal_closed_count": (
1 if retry_rollback_terminal.get("closed") is True else 0
),
"historical_stdin_boundary_replay_closed_count": (
1
if stdin_boundary_replay.get("latest_replay_attempt_closed")
is True
else 0
),
"historical_stdin_boundary_replay_runtime_apply_count": (
1
if (
stdin_boundary_replay.get("operation_boundaries") or {}
).get("runtime_apply_executed_by_replay")
is True
else 0
),
"live_mcp_context_count": _int_value(readback.get("mcp_context", {}).get("total")),
"live_service_log_evidence_count": _int_value(
readback.get("service_log_evidence", {}).get("total")
@@ -6604,10 +6809,42 @@ async def _load_ai_agent_autonomous_runtime_receipt_readback_uncached(
"retry_terminal_latest",
_RUNTIME_RETRY_TERMINAL_LATEST_SQL,
)
primary_apply_chain_anchor = await _safe_rows(
"primary_apply_chain_anchor",
_RUNTIME_PRIMARY_APPLY_CHAIN_ANCHOR_SQL,
)
operation_latest = await _safe_rows(
"operation_latest",
_RUNTIME_OPERATION_LATEST_SQL,
)
if primary_apply_chain_anchor:
anchor = _row_mapping(primary_apply_chain_anchor[0])
anchor_refs = list(dict.fromkeys(
str(anchor.get(key) or "")
for key in (
"apply_op_id",
"check_mode_op_id",
"candidate_op_id",
)
if str(anchor.get(key) or "")
))
if anchor_refs:
padded_refs = [*anchor_refs[:3], "", "", ""][:3]
params.update(
{
"operation_chain_ref_1": padded_refs[0],
"operation_chain_ref_2": padded_refs[1],
"operation_chain_ref_3": padded_refs[2],
}
)
primary_apply_chain = await _safe_rows(
"primary_apply_chain",
_RUNTIME_OPERATION_CHAIN_SQL,
)
operation_latest = _merge_runtime_operation_rows(
primary_apply_chain,
operation_latest,
)
for chain_pass in range(2):
missing_chain_refs = _missing_runtime_operation_chain_ref_ids(
operation_latest
@@ -7392,6 +7629,19 @@ _RUNTIME_OPERATION_COUNTS_DIRECT_SQL = """
"""
_RUNTIME_PRIMARY_APPLY_CHAIN_ANCHOR_SQL = """
SELECT
op_id::text AS apply_op_id,
parent_op_id::text AS check_mode_op_id,
input ->> 'source_candidate_op_id' AS candidate_op_id
FROM automation_operation_log
WHERE operation_type = 'ansible_apply_executed'
AND status IN ('success', 'failed')
ORDER BY created_at DESC
LIMIT 1
"""
_RUNTIME_OPERATION_LATEST_SQL = """
WITH latest_apply_chain AS (
SELECT
@@ -7424,6 +7674,12 @@ _RUNTIME_OPERATION_LATEST_SQL = """
input ->> 'catalog_id' AS catalog_id,
coalesce(input ->> 'apply_playbook_path', input ->> 'playbook_path') AS playbook_path,
input ->> 'execution_mode' AS execution_mode,
input ->> 'historical_stdin_boundary_replay'
AS historical_stdin_boundary_replay,
input ->> 'replay_of_check_mode_op_id'
AS replay_of_check_mode_op_id,
input ->> 'retry_of_check_mode_op_id'
AS retry_of_check_mode_op_id,
input ->> 'source_candidate_op_id' AS source_candidate_op_id,
input ->> 'check_mode_op_id' AS check_mode_op_id,
input ->> 'risk_level' AS risk_level,
@@ -7495,6 +7751,9 @@ _RUNTIME_OPERATION_LATEST_SQL = """
OR operation_row.input ->> 'automation_run_id'
= latest_apply_chain.candidate_op_id
THEN 0
WHEN operation_row.input ->> 'execution_mode'
= 'stdin_boundary_check_mode_replay_terminal'
THEN 1
WHEN coalesce(
input ->> 'semantic_operation_type',
operation_type
@@ -7539,6 +7798,12 @@ _RUNTIME_OPERATION_LATEST_DIRECT_SQL = """
input ->> 'catalog_id' AS catalog_id,
coalesce(input ->> 'apply_playbook_path', input ->> 'playbook_path') AS playbook_path,
input ->> 'execution_mode' AS execution_mode,
input ->> 'historical_stdin_boundary_replay'
AS historical_stdin_boundary_replay,
input ->> 'replay_of_check_mode_op_id'
AS replay_of_check_mode_op_id,
input ->> 'retry_of_check_mode_op_id'
AS retry_of_check_mode_op_id,
input ->> 'source_candidate_op_id' AS source_candidate_op_id,
input ->> 'check_mode_op_id' AS check_mode_op_id,
input ->> 'risk_level' AS risk_level,
@@ -7607,6 +7872,9 @@ _RUNTIME_OPERATION_LATEST_DIRECT_SQL = """
OR operation_row.input ->> 'automation_run_id'
= latest_apply_chain.candidate_op_id
THEN 0
WHEN operation_row.input ->> 'execution_mode'
= 'stdin_boundary_check_mode_replay_terminal'
THEN 1
WHEN coalesce(
input ->> 'semantic_operation_type',
operation_type
@@ -7916,6 +8184,12 @@ _RUNTIME_OPERATION_CHAIN_SQL = """
input ->> 'catalog_id' AS catalog_id,
coalesce(input ->> 'apply_playbook_path', input ->> 'playbook_path') AS playbook_path,
input ->> 'execution_mode' AS execution_mode,
input ->> 'historical_stdin_boundary_replay'
AS historical_stdin_boundary_replay,
input ->> 'replay_of_check_mode_op_id'
AS replay_of_check_mode_op_id,
input ->> 'retry_of_check_mode_op_id'
AS retry_of_check_mode_op_id,
input ->> 'source_candidate_op_id' AS source_candidate_op_id,
input ->> 'check_mode_op_id' AS check_mode_op_id,
input ->> 'risk_level' AS risk_level,
@@ -7944,10 +8218,10 @@ _RUNTIME_OPERATION_CHAIN_SQL = """
duration_ms,
created_at
FROM automation_operation_log
WHERE op_id::text IN (
:operation_chain_ref_1,
:operation_chain_ref_2,
:operation_chain_ref_3
WHERE op_id IN (
CAST(NULLIF(:operation_chain_ref_1, '') AS uuid),
CAST(NULLIF(:operation_chain_ref_2, '') AS uuid),
CAST(NULLIF(:operation_chain_ref_3, '') AS uuid)
)
ORDER BY created_at DESC
"""
@@ -7971,6 +8245,12 @@ _RUNTIME_OPERATION_CHAIN_DIRECT_SQL = """
input ->> 'catalog_id' AS catalog_id,
coalesce(input ->> 'apply_playbook_path', input ->> 'playbook_path') AS playbook_path,
input ->> 'execution_mode' AS execution_mode,
input ->> 'historical_stdin_boundary_replay'
AS historical_stdin_boundary_replay,
input ->> 'replay_of_check_mode_op_id'
AS replay_of_check_mode_op_id,
input ->> 'retry_of_check_mode_op_id'
AS retry_of_check_mode_op_id,
input ->> 'source_candidate_op_id' AS source_candidate_op_id,
input ->> 'check_mode_op_id' AS check_mode_op_id,
input ->> 'risk_level' AS risk_level,
@@ -7999,7 +8279,11 @@ _RUNTIME_OPERATION_CHAIN_DIRECT_SQL = """
duration_ms,
created_at
FROM automation_operation_log
WHERE op_id::text IN ($1, $2, $3)
WHERE op_id IN (
NULLIF($1, '')::uuid,
NULLIF($2, '')::uuid,
NULLIF($3, '')::uuid
)
ORDER BY created_at DESC
"""

110
apps/api/src/services/telegram_alert_monitoring_coverage_readback.py
View File

@@ -72,13 +72,113 @@ _RUNTIME_LIFECYCLE_SQL = """
) AS started_count,
COUNT(*) FILTER (
WHERE event_type = 'EXECUTION_COMPLETED'
AND success IS TRUE
) AS completed_success_count,
AND (
success IS TRUE
OR (
success IS FALSE
AND EXISTS (
SELECT 1
FROM automation_operation_log apply
WHERE apply.op_id = CASE
WHEN lifecycle.context ->> 'apply_op_id'
~* '^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$'
THEN CAST(
lifecycle.context ->> 'apply_op_id'
AS uuid
)
ELSE NULL
END
AND apply.operation_type
= 'ansible_apply_executed'
AND apply.status = 'failed'
AND apply.input ->> 'automation_run_id'
= COALESCE(
NULLIF(
lifecycle.context
->> 'automation_run_id',
''
),
NULLIF(
lifecycle.context ->> 'run_id',
''
)
)
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(
COALESCE(
apply.input
-> 'runtime_stage_receipts',
'[]'::jsonb
)
) retry_receipt(value)
WHERE retry_receipt.value ->> 'stage_id'
= 'retry_or_rollback'
AND retry_receipt.value
->> 'durable_receipt' = 'true'
AND retry_receipt.value #>>
'{detail,runtime_apply_executed}'
= 'false'
AND retry_receipt.value #>>
'{detail,verified_no_write_terminal}'
= 'true'
AND retry_receipt.value #>>
'{detail,repository_readback_verified}'
= 'true'
)
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(
COALESCE(
apply.input
-> 'runtime_stage_receipts',
'[]'::jsonb
)
) closure_receipt(value)
WHERE closure_receipt.value ->> 'stage_id'
= 'incident_closure'
AND closure_receipt.value
->> 'durable_receipt' = 'true'
AND closure_receipt.value #>>
'{detail,no_write_terminal}'
= 'true'
AND closure_receipt.value #>>
'{detail,telegram_receipt_acknowledged}'
= 'true'
AND closure_receipt.value #>>
'{detail,repository_readback_verified}'
= 'true'
)
)
AND EXISTS (
SELECT 1
FROM alert_operation_log telegram_result
WHERE telegram_result.event_type
= 'TELEGRAM_RESULT_SENT'
AND telegram_result.context
->> 'automation_run_id'
= COALESCE(
NULLIF(
lifecycle.context
->> 'automation_run_id',
''
),
NULLIF(
lifecycle.context ->> 'run_id',
''
)
)
AND telegram_result.context ->> 'apply_op_id'
= lifecycle.context ->> 'apply_op_id'
)
)
)
) AS completed_terminal_count,
COUNT(*) FILTER (
WHERE event_type = 'TELEGRAM_RESULT_SENT'
) AS result_sent_count,
MAX(created_at) AS latest_at
FROM alert_operation_log
FROM alert_operation_log lifecycle
WHERE created_at >= NOW() - INTERVAL '7 days'
AND COALESCE(
NULLIF(context->>'automation_run_id', ''),
@@ -93,7 +193,7 @@ _RUNTIME_LIFECYCLE_SQL = """
COUNT(*) FILTER (
WHERE triggered_count > 0
AND started_count > 0
AND completed_success_count > 0
AND completed_terminal_count > 0
AND result_sent_count > 0
) AS closed_run_count,
MAX(latest_at) FILTER (
@@ -102,7 +202,7 @@ _RUNTIME_LIFECYCLE_SQL = """
MAX(latest_at) FILTER (
WHERE triggered_count > 0
AND started_count > 0
AND completed_success_count > 0
AND completed_terminal_count > 0
AND result_sent_count > 0
) AS latest_closed_at
FROM run_lifecycle