From afe15662f339660db9c8fc9270e27217358df311 Mon Sep 17 00:00:00 2001 From: ogt Date: Tue, 14 Jul 2026 11:24:26 +0800 Subject: [PATCH] fix(agent): keep alert replay closure monotonic --- .../ai_agent_autonomous_runtime_control.py | 294 +++++++++++++++++- ...gram_alert_monitoring_coverage_readback.py | 110 ++++++- ...est_ai_agent_autonomous_runtime_control.py | 221 +++++++++++++ ..._alert_monitoring_coverage_readback_api.py | 17 + apps/web/messages/en.json | 6 + apps/web/messages/zh-TW.json | 6 + .../autonomous-runtime-receipt-panel.tsx | 46 +++ 7 files changed, 690 insertions(+), 10 deletions(-) diff --git a/apps/api/src/services/ai_agent_autonomous_runtime_control.py b/apps/api/src/services/ai_agent_autonomous_runtime_control.py index 7d07dce23..f3b7f8cd4 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -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 """ diff --git a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py index fa428038f..659981f14 100644 --- a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py +++ b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py @@ -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 diff --git a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py index abc631753..02d9156c8 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -102,6 +102,68 @@ def _closed_retry_terminal_row() -> dict[str, object]: } +def _stdin_boundary_replay_rows() -> list[dict[str, object]]: + run_id = "historical-candidate-op" + failed_check_op_id = "original-failed-check-op" + replay_check_op_id = "stdin-replay-check-op" + terminal_op_id = "stdin-replay-terminal-op" + return [ + { + "op_id": terminal_op_id, + "parent_op_id": replay_check_op_id, + "operation_type": "ansible_execution_skipped", + "status": "failed", + "automation_run_id": run_id, + "execution_mode": "stdin_boundary_check_mode_replay_terminal", + "runtime_stage_receipts": [ + { + "schema_version": "ai_automation_stage_receipt_v1", + "stage_id": "retry_or_rollback", + "automation_run_id": run_id, + "durable_receipt": True, + "evidence_ref": ( + f"automation_operation_log:{terminal_op_id}:dry_run_result" + ), + "detail": { + "schema_version": ( + "ansible_failed_check_stdin_boundary_retry_terminal_v1" + ), + "failed_check_mode_op_id": failed_check_op_id, + "retry_check_mode_op_id": replay_check_op_id, + "retry_terminal_op_id": terminal_op_id, + "terminal_type": ( + "no_write_stdin_boundary_replay_failed_waiting_transport_repair" + ), + "check_mode_replay_performed": True, + "check_mode_replay_returncode": 4, + "runtime_apply_executed": False, + "verified_no_write_terminal": True, + "retry_operation_readback_verified": True, + "retry_attempt": 1, + "max_retry_attempts": 1, + "bounded_retry": True, + "repository_readback_verified": True, + "durable_write_acknowledged": True, + "safe_next_action": "queue_ai_transport_repair_candidate", + "raw_log_payload_stored": False, + "secret_value_stored": False, + }, + } + ], + }, + { + "op_id": replay_check_op_id, + "parent_op_id": run_id, + "operation_type": "ansible_check_mode_executed", + "status": "failed", + "automation_run_id": run_id, + "execution_mode": "stdin_boundary_check_mode_replay", + "historical_stdin_boundary_replay": "true", + "replay_of_check_mode_op_id": failed_check_op_id, + }, + ] + + class _FailingRuntimeDb: async def execute(self, *_args, **_kwargs): raise RuntimeError("runtime trace query unavailable") @@ -161,6 +223,69 @@ class _BudgetRuntimeDbContext: return False +class _PrimaryAnchorRuntimeDb: + def __init__(self): + self.statements: list[str] = [] + + async def execute(self, statement, *_args, **_kwargs): + sql = str(statement) + self.statements.append(sql) + if "SET LOCAL statement_timeout" in sql: + return _FakeMappingResult([]) + if "WITH latest_retry AS" in sql: + return _FakeMappingResult([]) + if ( + "op_id::text AS apply_op_id" in sql + and "WITH latest_apply_chain AS" not in sql + ): + return _FakeMappingResult([ + { + "apply_op_id": "stable-apply-op", + "check_mode_op_id": "stable-check-op", + "candidate_op_id": "stable-candidate-op", + } + ]) + if "WITH latest_apply_chain AS" in sql: + return _FakeMappingResult(_stdin_boundary_replay_rows()) + if "CAST(NULLIF(:operation_chain_ref_1" in sql: + return _FakeMappingResult([ + { + "op_id": "stable-apply-op", + "parent_op_id": "stable-check-op", + "operation_type": "ansible_apply_executed", + "status": "failed", + "automation_run_id": "stable-candidate-op", + "source_candidate_op_id": "stable-candidate-op", + "check_mode_op_id": "stable-check-op", + }, + { + "op_id": "stable-check-op", + "parent_op_id": "stable-candidate-op", + "operation_type": "ansible_check_mode_executed", + "status": "success", + "automation_run_id": "stable-candidate-op", + }, + { + "op_id": "stable-candidate-op", + "operation_type": "ansible_candidate_matched", + "status": "dry_run", + "automation_run_id": "stable-candidate-op", + }, + ]) + return _FakeMappingResult([]) + + +class _PrimaryAnchorRuntimeDbContext: + def __init__(self, db: _PrimaryAnchorRuntimeDb): + self._db = db + + async def __aenter__(self) -> _PrimaryAnchorRuntimeDb: + return self._db + + async def __aexit__(self, _exc_type, _exc, _tb) -> bool: + return False + + class _SlowEnterRuntimeDbContext: async def __aenter__(self): await asyncio.sleep(1) @@ -3255,6 +3380,99 @@ def test_runtime_operation_latest_queries_prioritize_latest_apply_chain() -> Non assert "latest_apply_chain.candidate_op_id" in sql assert "operation_row.operation_type IN" in sql assert "'remediation_executed'" in sql + assert "'stdin_boundary_check_mode_replay_terminal'" in sql + + +def test_runtime_primary_apply_anchor_is_global_and_chain_lookup_uses_uuid_index(): + anchor_sql = runtime_control_module._RUNTIME_PRIMARY_APPLY_CHAIN_ANCHOR_SQL + chain_sql = runtime_control_module._RUNTIME_OPERATION_CHAIN_SQL + + assert "operation_type = 'ansible_apply_executed'" in anchor_sql + assert "AND status IN ('success', 'failed')" in anchor_sql + assert "ORDER BY created_at DESC" in anchor_sql + assert "lookback_hours" not in anchor_sql + assert "WHERE op_id IN" in chain_sql + assert "CAST(NULLIF(:operation_chain_ref_1, '') AS uuid)" in chain_sql + assert "WHERE op_id::text IN" not in chain_sql + + +@pytest.mark.asyncio +async def test_runtime_loader_keeps_primary_apply_anchor_when_historical_replay_runs( + monkeypatch, +): + fake_db = _PrimaryAnchorRuntimeDb() + + async def _fake_consumer_readback(*, project_id: str): + assert project_id == "awoooi" + return _log_controlled_writeback_consumer_readback() + + monkeypatch.setattr( + runtime_control_module, + "get_db_context", + lambda project_id: _PrimaryAnchorRuntimeDbContext(fake_db), + ) + monkeypatch.setattr( + runtime_control_module, + "_load_log_controlled_writeback_consumer_readback", + _fake_consumer_readback, + ) + + readback = await ( + runtime_control_module._load_ai_agent_autonomous_runtime_receipt_readback_uncached() + ) + + ledger = readback["autonomous_execution_loop_ledger"] + assert ledger["automation_run_id"] == "stable-candidate-op" + assert ledger["apply_op_id"] == "stable-apply-op" + assert ledger["check_mode_op_id"] == "stable-check-op" + assert ledger["root_candidate_op_id"] == "stable-candidate-op" + + replay = readback["historical_stdin_boundary_replay"] + assert replay["latest_replay_attempt_closed"] is True + assert replay["status"] == "verified_no_write_terminal_repair_required" + assert replay["check_mode_replay_returncode"] == 4 + assert set(replay["controls"].values()) == {True} + assert replay["operation_boundaries"][ + "runtime_apply_executed_by_replay" + ] is False + assert replay["operation_boundaries"][ + "primary_runtime_flow_replaced" + ] is False + + control = build_ai_agent_autonomous_runtime_control() + runtime_control_module._attach_runtime_receipt_readback(control, readback) + assert control["historical_stdin_boundary_replay"] == replay + assert control["rollups"][ + "historical_stdin_boundary_replay_closed_count" + ] == 1 + assert control["rollups"][ + "historical_stdin_boundary_replay_runtime_apply_count" + ] == 0 + + +def test_stdin_boundary_replay_reports_unexpected_runtime_apply() -> None: + rows = _stdin_boundary_replay_rows() + receipts = rows[0]["runtime_stage_receipts"] + assert isinstance(receipts, list) + detail = receipts[0]["detail"] + assert isinstance(detail, dict) + detail["runtime_apply_executed"] = True + + readback = build_runtime_receipt_readback_from_rows( + operation_latest_rows=rows + ) + replay = readback["historical_stdin_boundary_replay"] + assert replay["latest_replay_attempt_closed"] is False + assert replay["controls"]["runtime_apply_not_executed"] is False + assert replay["operation_boundaries"][ + "runtime_apply_executed_by_replay" + ] is True + + control = build_ai_agent_autonomous_runtime_control() + runtime_control_module._attach_runtime_receipt_readback(control, readback) + assert control["rollups"][ + "historical_stdin_boundary_replay_runtime_apply_count" + ] == 1 def test_execution_capability_lifecycle_requires_same_run_terminal_receipt() -> None: @@ -3511,6 +3729,9 @@ def test_runtime_receipt_queries_prioritize_closure_and_bound_latest_window(): ) assert source.index('"retry_terminal_latest"') < source.index( + '"primary_apply_chain_anchor"' + ) + assert source.index('"primary_apply_chain_anchor"') < source.index( '"operation_latest"' ) assert source.index('"operation_latest"') < source.index( diff --git a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py index 99040e510..df562da36 100644 --- a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py +++ b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py @@ -800,6 +800,23 @@ def test_telegram_alert_runtime_log_calculates_incident_closure() -> None: assert summary["automation_runtime_closure_ready"] is False +def test_runtime_lifecycle_sql_accepts_only_durable_failed_terminals() -> None: + sql = coverage_service._RUNTIME_LIFECYCLE_SQL + + assert "completed_terminal_count" in sql + assert "apply.status = 'failed'" in sql + assert "= 'retry_or_rollback'" in sql + assert "'{detail,runtime_apply_executed}'" in sql + assert "'{detail,verified_no_write_terminal}'" in sql + assert "= 'incident_closure'" in sql + assert "'{detail,no_write_terminal}'" in sql + assert "'{detail,telegram_receipt_acknowledged}'" in sql + assert "'{detail,repository_readback_verified}'" in sql + assert "telegram_result.event_type" in sql + assert "= 'TELEGRAM_RESULT_SENT'" in sql + assert "completed_success_count" not in sql + + @pytest.mark.asyncio async def test_telegram_alert_monitoring_marks_ai_alert_cards_source_unavailable( monkeypatch: pytest.MonkeyPatch, diff --git a/apps/web/messages/en.json b/apps/web/messages/en.json index 8a521b830..76f0a909e 100644 --- a/apps/web/messages/en.json +++ b/apps/web/messages/en.json @@ -12117,6 +12117,12 @@ "degraded": "讀取降級", "unavailable": "讀取失敗" }, + "replay": { + "closed": "No-write replay closed", + "closedRepairRequired": "No-write closed · AI repair queued", + "pending": "Historical replay open", + "applyObserved": "Unexpected replay apply detected" + }, "metrics": { "loop": "Loop", "trace": "Trace", diff --git a/apps/web/messages/zh-TW.json b/apps/web/messages/zh-TW.json index 33a5d212d..dad6e3544 100644 --- a/apps/web/messages/zh-TW.json +++ b/apps/web/messages/zh-TW.json @@ -12117,6 +12117,12 @@ "degraded": "讀取降級", "unavailable": "讀取失敗" }, + "replay": { + "closed": "無寫入重放已封口", + "closedRepairRequired": "無寫入封口 · 待 AI 修復", + "pending": "歷史重放待封口", + "applyObserved": "重放偵測到異常 apply" + }, "metrics": { "loop": "Loop", "trace": "Trace", diff --git a/apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx b/apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx index 9121730ae..c6802f6b1 100644 --- a/apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx +++ b/apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx @@ -167,6 +167,15 @@ type RuntimeReceiptReadback = { incident_id?: string | null; catalog_id?: string | null; } | null; + historical_stdin_boundary_replay?: HistoricalStdinBoundaryReplay | null; +}; + +type HistoricalStdinBoundaryReplay = { + status?: string | null; + latest_replay_attempt_closed?: boolean | null; + operation_boundaries?: { + runtime_apply_executed_by_replay?: boolean | null; + } | null; }; type RuntimeControlPayload = { @@ -182,6 +191,7 @@ type RuntimeControlPayload = { critical_break_glass_required?: boolean | null; } | null; runtime_receipt_readback?: RuntimeReceiptReadback | null; + historical_stdin_boundary_replay?: HistoricalStdinBoundaryReplay | null; rollups?: Record | null; }; @@ -692,6 +702,26 @@ export function AutonomousRuntimeReceiptPanel({ const sourceFamilyItems = readback?.work_item_progress?.source_family_items ?? []; const latestFlow = readback?.latest_flow_closure; const rollups = payload?.rollups ?? {}; + const stdinBoundaryReplay = payload?.historical_stdin_boundary_replay + ?? readback?.historical_stdin_boundary_replay; + const replayApplyObserved = + stdinBoundaryReplay?.operation_boundaries?.runtime_apply_executed_by_replay === true + || toNumber(rollups.historical_stdin_boundary_replay_runtime_apply_count) > 0; + const replayClosed = stdinBoundaryReplay?.latest_replay_attempt_closed === true; + const replayRepairRequired = + stdinBoundaryReplay?.status === "verified_no_write_terminal_repair_required"; + const replayTone: Tone = replayApplyObserved + ? "warn" + : replayClosed + ? "ok" + : "warn"; + const replayLabel = replayApplyObserved + ? t("replay.applyObserved") + : replayClosed + ? replayRepairRequired + ? t("replay.closedRepairRequired") + : t("replay.closed") + : t("replay.pending"); const consumerRollups = consumerPayload?.rollups ?? {}; const consumerBlockers = consumerPayload?.active_blockers ?? []; const telegramVerifierRollups = telegramVerifierPayload?.rollups ?? {}; @@ -1536,6 +1566,22 @@ export function AutonomousRuntimeReceiptPanel({ {stateLabel} + {stdinBoundaryReplay ? ( + + {replayApplyObserved ? ( + + ) : null} {hasRuntimeReadback ? t("completion", {