From 5aa651563c6abb37ad69aadabcd238fe6307b52e Mon Sep 17 00:00:00 2001 From: ogt Date: Tue, 14 Jul 2026 13:18:00 +0800 Subject: [PATCH] fix(agent): close no-write alert replay receipts --- .../ai_agent_autonomous_runtime_control.py | 42 + .../services/awooop_ansible_audit_service.py | 157 ++ .../awooop_ansible_check_mode_service.py | 1286 ++++++++++++++++- .../awooop_ansible_learning_writeback.py | 103 +- ...gram_alert_monitoring_coverage_readback.py | 73 +- apps/api/src/services/telegram_gateway.py | 32 +- ...est_ai_agent_autonomous_runtime_control.py | 46 + .../tests/test_ansible_verified_closure.py | 6 + .../tests/test_awooop_truth_chain_service.py | 286 +++- ..._signal_worker_ansible_executor_binding.py | 1 + ..._alert_monitoring_coverage_readback_api.py | 19 + .../tests/test_telegram_message_templates.py | 73 + k8s/awoooi-prod/08-deployment-worker.yaml | 2 +- 13 files changed, 2088 insertions(+), 38 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 f3b7f8cd4..d9f8884f0 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -7235,6 +7235,10 @@ async def build_ai_agent_autonomous_runtime_control_with_live_readback( ) _attach_runtime_receipt_readback(payload, readback) payload["executor_trust_boundary"] = boundary_readback + _attach_effective_executor_runtime_switches( + payload, + boundary_readback=boundary_readback, + ) _attach_autonomous_single_writer_readback( payload, readback=readback, @@ -7257,6 +7261,44 @@ async def build_ai_agent_autonomous_runtime_control_with_live_readback( return payload +def _attach_effective_executor_runtime_switches( + payload: dict[str, Any], + *, + boundary_readback: Mapping[str, Any], +) -> None: + """Separate the API process flag from the dedicated broker truth.""" + + runtime_switches = payload.setdefault("runtime_switches", {}) + controls = boundary_readback.get("controls") + if not isinstance(controls, Mapping): + controls = {} + dedicated_broker_ready = bool( + boundary_readback.get("status") == "verified_ready" + and boundary_readback.get("production_boundary_verified") is True + and boundary_readback.get("source_sha_matches_deployment") is True + and controls.get("broker_ssh_mount_present") is True + and controls.get("broker_ssh_egress_allowlisted") is True + and controls.get("broker_kubernetes_token_absent") is True + ) + runtime_switches.update( + { + "ansible_check_mode_worker_scope": "api_process_only", + "dedicated_execution_broker_ready": dedicated_broker_ready, + "effective_ansible_executor_available": dedicated_broker_ready, + "effective_executor_readback_source": ( + boundary_readback.get("receipt_ref") + if dedicated_broker_ready + else None + ), + "effective_executor_verified_source_sha": ( + boundary_readback.get("verified_source_sha") + if dedicated_broker_ready + else None + ), + } + ) + + def _attach_autonomous_single_writer_readback( payload: dict[str, Any], *, diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index 308dda8b4..b28cdfdfd 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -178,6 +178,163 @@ def verified_ansible_candidate_terminal_sql(candidate_alias: str) -> str: = __AUTOMATION_RUN_ID__ ) ) + OR EXISTS ( + SELECT 1 + FROM automation_operation_log failed_check + JOIN automation_operation_log replay_check + ON replay_check.parent_op_id = failed_check.parent_op_id + AND replay_check.operation_type = 'ansible_check_mode_executed' + AND replay_check.status IN ('success', 'failed') + AND replay_check.input ->> 'execution_mode' + = 'stdin_boundary_check_mode_replay' + AND replay_check.input ->> 'replay_of_check_mode_op_id' + = failed_check.op_id::text + JOIN automation_operation_log terminal + ON terminal.parent_op_id = replay_check.op_id + AND terminal.operation_type = 'ansible_execution_skipped' + AND terminal.status IN ('success', 'failed') + AND terminal.input ->> 'execution_mode' + = 'stdin_boundary_check_mode_replay_terminal' + AND terminal.output ->> 'runtime_apply_executed' = 'false' + WHERE failed_check.parent_op_id = __CANDIDATE__.op_id + AND failed_check.operation_type = 'ansible_check_mode_executed' + AND failed_check.status = 'failed' + AND failed_check.input ->> 'automation_run_id' + = __AUTOMATION_RUN_ID__ + AND coalesce( + failed_check.incident_id::text, + failed_check.input ->> 'incident_id' + ) = __INCIDENT_ID__ + AND replay_check.input ->> 'automation_run_id' + = __AUTOMATION_RUN_ID__ + AND replay_check.input ->> 'replay_of_check_mode_op_id' + = failed_check.op_id::text + AND coalesce( + replay_check.incident_id::text, + replay_check.input ->> 'incident_id' + ) = __INCIDENT_ID__ + AND terminal.input ->> 'automation_run_id' + = __AUTOMATION_RUN_ID__ + AND terminal.input ->> 'retry_of_check_mode_op_id' + = failed_check.op_id::text + AND coalesce( + terminal.incident_id::text, + terminal.input ->> 'incident_id' + ) = __INCIDENT_ID__ + AND ( + SELECT COUNT(DISTINCT receipt.value ->> 'stage_id') + FROM jsonb_array_elements( + coalesce( + terminal.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) receipt(value) + WHERE receipt.value ->> 'stage_id' IN ( + 'mcp_context', + 'service_log_evidence', + 'normalized_asset_identity', + 'source_truth_diff', + 'risk_policy_decision', + 'executor_log_projection', + 'retry_or_rollback', + 'km_playbook_writeback', + 'rag_writeback', + 'playbook_trust', + 'timeline_projection' + ) + AND receipt.value ->> 'automation_run_id' + = __AUTOMATION_RUN_ID__ + AND receipt.value ->> 'durable_receipt' = 'true' + ) = 11 + AND EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + terminal.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) receipt(value) + WHERE receipt.value ->> 'stage_id' = 'retry_or_rollback' + AND receipt.value #>> + '{detail,retry_terminal_op_id}' + = terminal.op_id::text + AND receipt.value #>> + '{detail,verified_no_write_terminal}' = 'true' + AND receipt.value #>> + '{detail,runtime_apply_executed}' = 'false' + AND receipt.value #>> + '{detail,repository_readback_verified}' = 'true' + AND receipt.value ->> 'durable_receipt' = 'true' + ) + AND EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + terminal.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) receipt(value) + WHERE receipt.value ->> 'stage_id' = 'incident_closure' + AND receipt.value #>> '{detail,terminal_op_id}' + = terminal.op_id::text + AND receipt.value #>> '{detail,no_write_terminal}' + = 'true' + AND receipt.value #>> + '{detail,telegram_receipt_acknowledged}' = 'true' + AND receipt.value #>> + '{detail,learning_writeback_acknowledged}' = 'true' + AND receipt.value #>> + '{detail,incident_projection_readback_verified}' + = 'true' + AND receipt.value #>> + '{detail,projection_readback_verified}' = 'true' + AND receipt.value #>> + '{detail,repository_readback_verified}' = 'true' + AND receipt.value ->> 'durable_receipt' = 'true' + ) + AND EXISTS ( + SELECT 1 + FROM awooop_outbound_message outbound + WHERE outbound.channel_type = 'telegram' + AND outbound.send_status = 'sent' + AND outbound.provider_message_id IS NOT NULL + AND outbound.source_envelope #>> + '{callback_reply,action}' + = 'controlled_apply_result' + AND outbound.source_envelope #>> + '{callback_reply,execution_kind}' + = 'no_write_replay' + AND outbound.source_envelope #>> + '{callback_reply,automation_run_id}' + = __AUTOMATION_RUN_ID__ + AND outbound.source_envelope #>> + '{callback_reply,incident_id}' = __INCIDENT_ID__ + AND outbound.source_envelope #>> + '{callback_reply,apply_op_id}' + = terminal.op_id::text + ) + AND EXISTS ( + SELECT 1 + FROM alert_operation_log lifecycle + WHERE lifecycle.incident_id = __INCIDENT_ID__ + AND lifecycle.event_type::text = 'EXECUTION_COMPLETED' + AND lifecycle.success IS FALSE + AND lifecycle.context ->> 'apply_op_id' + = terminal.op_id::text + AND lifecycle.context ->> 'automation_run_id' + = __AUTOMATION_RUN_ID__ + ) + AND EXISTS ( + SELECT 1 + FROM alert_operation_log lifecycle + WHERE lifecycle.incident_id = __INCIDENT_ID__ + AND lifecycle.event_type::text = 'TELEGRAM_RESULT_SENT' + AND lifecycle.context ->> 'apply_op_id' + = terminal.op_id::text + AND lifecycle.context ->> 'automation_run_id' + = __AUTOMATION_RUN_ID__ + ) + ) """ return ( template.replace("__CANDIDATE__", candidate) diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 789890289..a83fe1465 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -79,6 +79,7 @@ _VERIFIED_CLOSURE_REQUIRED_STAGE_IDS = frozenset( "timeline_projection", } ) +_NO_WRITE_REPLAY_REQUIRED_STAGE_IDS = _VERIFIED_CLOSURE_REQUIRED_STAGE_IDS FORCED_COMMAND_BLOCKER = "ansible_repair_ssh_forced_command_denies_ansible_bootstrap" REPAIR_FORCED_COMMAND_KEY_PATH = Path("/etc/repair-ssh/id_ed25519") REPAIR_FORCED_COMMAND_KNOWN_HOSTS_PATH = Path("/etc/repair-known-hosts/known_hosts") @@ -1982,6 +1983,21 @@ async def _append_runtime_stage_receipts_to_apply( apply_op_id: str, receipts: tuple[dict[str, Any], ...], project_id: str, +) -> bool: + return await _append_runtime_stage_receipts_to_operation( + operation_id=apply_op_id, + operation_type="ansible_apply_executed", + receipts=receipts, + project_id=project_id, + ) + + +async def _append_runtime_stage_receipts_to_operation( + *, + operation_id: str, + operation_type: str, + receipts: tuple[dict[str, Any], ...], + project_id: str, ) -> bool: if not receipts: return False @@ -2021,19 +2037,21 @@ async def _append_runtime_stage_receipts_to_apply( ), true ) - WHERE apply.op_id = CAST(:apply_op_id AS uuid) - AND apply.operation_type = 'ansible_apply_executed' + WHERE apply.op_id = CAST(:operation_id AS uuid) + AND apply.operation_type = :operation_type """), { "receipts": json.dumps(receipts, ensure_ascii=False), - "apply_op_id": apply_op_id, + "operation_id": operation_id, + "operation_type": operation_type, }, ) return bool(updated.rowcount) except Exception as exc: logger.warning( "ansible_runtime_terminal_receipt_append_failed", - apply_op_id=apply_op_id, + apply_op_id=operation_id, + operation_type=operation_type, error_type=type(exc).__name__, ) return False @@ -3559,6 +3577,1216 @@ async def _record_stdin_boundary_replay_terminal( return False +def _claim_from_stdin_boundary_terminal_row( + row: dict[str, Any], +) -> AnsibleCheckModeClaim | None: + replay_input = _json_loads(row.get("replay_input")) + base = _claim_from_stale_check_mode_row( + { + "op_id": row.get("replay_op_id"), + "parent_op_id": row.get("source_candidate_op_id"), + "incident_id": row.get("incident_id"), + "input": replay_input, + } + ) + if base is None: + return None + terminal_input = _json_loads(row.get("terminal_input")) + automation_run_id = str( + terminal_input.get("automation_run_id") + or replay_input.get("automation_run_id") + or base.source_candidate_op_id + ) + return replace( + base, + input_payload={ + **base.input_payload, + **replay_input, + "automation_run_id": automation_run_id, + "historical_stdin_boundary_replay": True, + "controlled_apply_allowed": False, + "apply_enabled": False, + }, + ) + + +async def _record_no_write_replay_timeline_receipt( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + terminal_op_id: str, + project_id: str, +) -> dict[str, Any] | None: + automation_run_id = _automation_run_id_for_claim(claim) + description = ( + f"automation_run_id={automation_run_id};" + f"terminal_op_id={terminal_op_id};catalog_id={claim.catalog_id};" + f"returncode={result.returncode};runtime_apply_executed=false;" + "terminal=no_write_stdin_boundary_replay" + ) + try: + async with get_db_context(project_id) as db: + selected = await db.execute( + text(""" + WITH existing AS ( + SELECT id + FROM timeline_events + WHERE incident_id = :incident_id + AND actor = 'ansible_stdin_boundary_replay_worker' + AND description = :description + ORDER BY created_at DESC + LIMIT 1 + ), inserted AS ( + INSERT INTO timeline_events ( + id, incident_id, event_type, status, title, + description, actor, actor_role, risk_level, + created_at + ) + SELECT + gen_random_uuid()::text, + :incident_id, + 'exec', + 'error', + :title, + :description, + 'ansible_stdin_boundary_replay_worker', + 'ai_agent', + :risk_level, + NOW() + WHERE NOT EXISTS (SELECT 1 FROM existing) + RETURNING id + ) + SELECT id FROM inserted + UNION ALL + SELECT id FROM existing + LIMIT 1 + """), + { + "incident_id": claim.incident_id, + "title": ( + "AI no-write replay terminal: " + f"{claim.catalog_id}" + )[:500], + "description": description, + "risk_level": str(claim.risk_level or "")[:20] or None, + }, + ) + timeline_event_id = str(selected.scalar() or "") + except Exception as exc: + logger.warning( + "ansible_stdin_boundary_timeline_write_failed", + automation_run_id=automation_run_id, + terminal_op_id=terminal_op_id, + error_type=type(exc).__name__, + ) + return None + if not timeline_event_id: + return None + return _runtime_stage_receipt( + claim, + stage_id="timeline_projection", + evidence_ref=f"timeline_events:{timeline_event_id}", + detail={ + "timeline_event_id": timeline_event_id, + "terminal_op_id": terminal_op_id, + "returncode": result.returncode, + "runtime_apply_executed": False, + "projection_status": "repair_required", + "repository_readback_verified": True, + }, + derived_from_durable_chain=True, + ) + + +async def _record_no_write_replay_learning_receipts( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + terminal_op_id: str, + project_id: str, +) -> dict[str, Any]: + from src.models.knowledge import ( + EntrySource, + EntryStatus, + EntryType, + KnowledgeEntryCreate, + ) + from src.repositories.knowledge_repository import KnowledgeDBRepository + + automation_run_id = _automation_run_id_for_claim(claim) + canonical_playbook_id = canonical_ansible_playbook_id(claim.catalog_id) + path_type = f"ansible_no_write_replay:{terminal_op_id[:8]}" + receipts = await _load_pre_decision_context_runtime_stage_receipts( + claim, + project_id=project_id, + derived_from_durable_chain=True, + ) + failed_check_mode_op_id = str( + claim.input_payload.get("replay_of_check_mode_op_id") or "" + ) + if failed_check_mode_op_id: + receipts.append( + _build_stdin_boundary_replay_stage_receipt( + claim, + result, + failed_check_mode_op_id=failed_check_mode_op_id, + terminal_op_id=terminal_op_id, + ) + ) + receipts.extend( + [ + _runtime_stage_receipt( + claim, + stage_id="normalized_asset_identity", + evidence_ref=f"ansible_catalog:{claim.catalog_id}", + detail={ + "asset_kind": "managed_host", + "canonical_asset_ids": list(claim.inventory_hosts), + "catalog_id": claim.catalog_id, + "playbook_path": claim.playbook_path, + }, + derived_from_durable_chain=True, + ), + _runtime_stage_receipt( + claim, + stage_id="source_truth_diff", + evidence_ref=( + f"automation_operation_log:{claim.op_id}:dry_run_result" + ), + detail={ + "check_mode_op_id": claim.op_id, + "check_mode_replay_performed": True, + "check_mode_replay_returncode": result.returncode, + "diff": True, + "runtime_apply_executed": False, + }, + derived_from_durable_chain=True, + ), + _runtime_stage_receipt( + claim, + stage_id="risk_policy_decision", + evidence_ref=f"ansible_policy:{claim.catalog_id}", + detail={ + "risk_level": str(claim.risk_level or "").lower(), + "controlled_apply_allowed": ( + str(claim.risk_level or "").lower() + in {"low", "medium", "high"} + ), + "controlled_apply_executed": False, + "execution_scope": "historical_no_write_replay", + "apply_deferred_reason": ( + "stdin_boundary_transport_or_playbook_repair_required" + ), + "decision": ( + "verified_no_write_terminal_pending_ai_repair" + ), + "policy": "global_product_governance_v2", + }, + derived_from_durable_chain=True, + ), + _runtime_stage_receipt( + claim, + stage_id="executor_log_projection", + evidence_ref=( + f"automation_operation_log:{terminal_op_id}:output" + ), + detail={ + "terminal_op_id": terminal_op_id, + "returncode": result.returncode, + "duration_ms": result.duration_ms, + "runtime_apply_executed": False, + "projection_sanitized": True, + }, + derived_from_durable_chain=True, + ), + ] + ) + + km_writeback: dict[str, Any] | None = None + try: + async with get_db_context(project_id) as db: + existing = await db.execute( + text(""" + SELECT id + FROM knowledge_entries + WHERE related_incident_id = :incident_id + AND path_type = :path_type + LIMIT 1 + """), + { + "incident_id": claim.incident_id, + "path_type": path_type, + }, + ) + if existing.scalar() is None: + repo = KnowledgeDBRepository(db) + await repo.create( + KnowledgeEntryCreate( + title=f"AI 無寫入重試反模式：{claim.incident_id}", + content=( + "AI Agent 已完成歷史 Ansible stdin boundary " + "check-mode 重試，未執行 runtime apply。\n\n" + f"- Automation run: {automation_run_id}\n" + f"- Incident: {claim.incident_id}\n" + f"- Catalog: {claim.catalog_id}\n" + f"- PlayBook: {claim.playbook_path}\n" + f"- Terminal operation: {terminal_op_id}\n" + f"- Check-mode return code: {result.returncode}\n" + "- Runtime apply executed: false\n" + "- Next action: AI transport or PlayBook repair\n" + ), + entry_type=EntryType.ANTI_PATTERN, + category="AI自動化/Ansible反模式", + tags=[ + "ai_auto_repair", + "ansible_no_write_replay", + "stdin_boundary", + str(claim.catalog_id or ""), + f"automation_run_id:{automation_run_id}", + ], + source=EntrySource.AI_EXTRACTED, + status=EntryStatus.PUBLISHED, + related_incident_id=claim.incident_id, + related_playbook_id=canonical_playbook_id, + path_type=path_type, + created_by="ai_agent_ansible_replay_worker", + ) + ) + readback = await db.execute( + text(""" + SELECT id, related_playbook_id, path_type, status, updated_at + FROM knowledge_entries + WHERE related_incident_id = :incident_id + AND path_type = :path_type + LIMIT 1 + """), + { + "incident_id": claim.incident_id, + "path_type": path_type, + }, + ) + km_row = readback.mappings().first() + if ( + km_row + and str(km_row.get("related_playbook_id") or "") + == canonical_playbook_id + and str(km_row.get("status") or "").lower() == "published" + and km_row.get("updated_at") is not None + ): + km_writeback = { + "schema_version": "ansible_no_write_km_writeback_v1", + "knowledge_entry_id": str(km_row["id"]), + "canonical_playbook_id": canonical_playbook_id, + "path_type": path_type, + "status": str(km_row.get("status") or ""), + "repository_write_acknowledged": True, + "repository_readback_verified": True, + "durable_write_acknowledged": True, + "raw_log_payload_stored": False, + "secret_value_stored": False, + } + except Exception as exc: + logger.warning( + "ansible_stdin_boundary_km_writeback_failed", + automation_run_id=automation_run_id, + terminal_op_id=terminal_op_id, + error_type=type(exc).__name__, + ) + if km_writeback is not None: + receipts.append( + _runtime_stage_receipt( + claim, + stage_id="km_playbook_writeback", + evidence_ref=( + "knowledge_entries:" + f"{km_writeback['knowledge_entry_id']}:row_version" + ), + detail=km_writeback, + derived_from_durable_chain=True, + ) + ) + + trust_writeback = await record_ansible_playbook_trust_writeback( + project_id=project_id, + automation_run_id=automation_run_id, + incident_id=claim.incident_id, + incident_db_id=_automation_operation_log_incident_id( + claim.incident_id + ), + catalog_id=claim.catalog_id, + playbook_path=claim.playbook_path, + apply_op_id=terminal_op_id, + verification_result="not_applicable_no_runtime_apply", + trust_mutation_performed=False, + observation_kind="no_write_replay", + ) + if trust_writeback is not None: + receipts.append( + _runtime_stage_receipt( + claim, + stage_id="playbook_trust", + evidence_ref=( + "playbooks:" + f"{trust_writeback['canonical_playbook_id']}:trust" + ), + detail=trust_writeback, + derived_from_durable_chain=True, + ) + ) + + rag_writeback = ( + await ensure_ansible_rag_writeback( + project_id=project_id, + incident_id=claim.incident_id, + apply_op_id=terminal_op_id, + path_type=path_type, + ) + if km_writeback is not None + else None + ) + if rag_writeback is not None: + rag_evidence_ref = ( + "knowledge_entries:" + f"{rag_writeback['knowledge_entry_id']}:embedding" + if rag_writeback.get("embedding_persisted") is True + else ( + "rag_chunks:" + f"{rag_writeback.get('source_id', '')}:" + f"count={rag_writeback.get('rag_chunk_count', 0)}" + ) + ) + receipts.append( + _runtime_stage_receipt( + claim, + stage_id="rag_writeback", + evidence_ref=rag_evidence_ref, + detail=rag_writeback, + derived_from_durable_chain=True, + ) + ) + + timeline_receipt = await _record_no_write_replay_timeline_receipt( + claim, + result, + terminal_op_id=terminal_op_id, + project_id=project_id, + ) + if timeline_receipt is not None: + receipts.append(timeline_receipt) + + by_stage = { + str(receipt.get("stage_id") or ""): receipt + for receipt in receipts + if isinstance(receipt, Mapping) + } + stage_ids = set(by_stage) + missing_stage_ids = sorted( + _NO_WRITE_REPLAY_REQUIRED_STAGE_IDS - stage_ids + ) + return { + "ready": not missing_stage_ids, + "receipts": tuple(by_stage.values()), + "stage_ids": sorted(stage_ids), + "missing_stage_ids": missing_stage_ids, + "km_writeback": km_writeback is not None, + "rag_writeback": rag_writeback is not None, + "playbook_trust": trust_writeback is not None, + "mcp_context": "mcp_context" in stage_ids, + "runtime_apply_executed": False, + } + + +async def _load_missing_stdin_boundary_terminal_projection_rows( + *, + project_id: str, + window_hours: int, + limit: int, +) -> list[dict[str, Any]]: + async with get_db_context(project_id) as db: + await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) + result = await db.execute( + text(""" + SELECT + terminal.op_id::text AS terminal_op_id, + terminal.status AS terminal_status, + terminal.input AS terminal_input, + terminal.output AS terminal_output, + terminal.error AS terminal_error, + terminal.duration_ms AS terminal_duration_ms, + replay.op_id::text AS replay_op_id, + candidate.op_id::text AS source_candidate_op_id, + replay.input AS replay_input, + coalesce( + terminal.incident_id::text, + terminal.input ->> 'incident_id', + replay.incident_id::text, + replay.input ->> 'incident_id' + ) AS incident_id + FROM automation_operation_log terminal + JOIN automation_operation_log replay + ON replay.op_id = terminal.parent_op_id + AND replay.operation_type = 'ansible_check_mode_executed' + AND replay.input ->> 'execution_mode' + = 'stdin_boundary_check_mode_replay' + AND replay.input ->> 'automation_run_id' + = terminal.input ->> 'automation_run_id' + JOIN automation_operation_log failed_check + ON failed_check.op_id::text + = replay.input ->> 'replay_of_check_mode_op_id' + AND failed_check.operation_type + = 'ansible_check_mode_executed' + AND failed_check.status = 'failed' + AND failed_check.parent_op_id = replay.parent_op_id + AND failed_check.input ->> 'automation_run_id' + = terminal.input ->> 'automation_run_id' + JOIN automation_operation_log candidate + ON candidate.op_id = replay.parent_op_id + AND candidate.operation_type = 'ansible_candidate_matched' + AND coalesce( + nullif(candidate.input ->> 'automation_run_id', ''), + candidate.op_id::text + ) = terminal.input ->> 'automation_run_id' + JOIN incidents incident + ON incident.incident_id = coalesce( + terminal.incident_id::text, + terminal.input ->> 'incident_id', + replay.incident_id::text, + replay.input ->> 'incident_id' + ) + AND incident.project_id = :project_id + WHERE terminal.operation_type = 'ansible_execution_skipped' + AND terminal.status IN ('success', 'failed') + AND terminal.input ->> 'execution_mode' + = 'stdin_boundary_check_mode_replay_terminal' + AND terminal.input ->> 'retry_of_check_mode_op_id' + = failed_check.op_id::text + AND terminal.output ->> 'runtime_apply_executed' = 'false' + AND terminal.created_at >= NOW() - ( + :window_hours * INTERVAL '1 hour' + ) + AND EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + terminal.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) receipt(value) + WHERE receipt.value ->> 'stage_id' + = 'retry_or_rollback' + AND receipt.value ->> 'automation_run_id' + = terminal.input ->> 'automation_run_id' + AND receipt.value #>> + '{detail,verified_no_write_terminal}' = 'true' + AND receipt.value #>> + '{detail,runtime_apply_executed}' = 'false' + AND receipt.value #>> + '{detail,repository_readback_verified}' = 'true' + AND receipt.value ->> 'durable_receipt' = 'true' + ) + AND NOT EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + terminal.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) receipt(value) + WHERE receipt.value ->> 'stage_id' = 'incident_closure' + AND receipt.value ->> 'automation_run_id' + = terminal.input ->> 'automation_run_id' + AND receipt.value #>> + '{detail,terminal_op_id}' = terminal.op_id::text + AND receipt.value #>> + '{detail,projection_readback_verified}' = 'true' + AND receipt.value ->> 'durable_receipt' = 'true' + ) + ORDER BY terminal.created_at ASC, terminal.op_id ASC + LIMIT :limit + """), + { + "project_id": project_id, + "window_hours": max(1, min(window_hours, 7 * 24)), + "limit": max(1, min(limit, 5)), + }, + ) + return [dict(row) for row in result.mappings().all()] + + +async def _stdin_boundary_replay_telegram_receipt_acknowledged( + claim: AnsibleCheckModeClaim, + *, + terminal_op_id: str, + project_id: str, +) -> bool: + automation_run_id = _automation_run_id_for_claim(claim) + try: + async with get_db_context(project_id) as db: + result = await db.execute( + text(""" + SELECT EXISTS ( + SELECT 1 + FROM awooop_outbound_message outbound + WHERE outbound.project_id = :project_id + AND outbound.channel_type = 'telegram' + AND outbound.send_status = 'sent' + AND outbound.provider_message_id IS NOT NULL + AND outbound.source_envelope #>> + '{callback_reply,action}' + = 'controlled_apply_result' + AND coalesce( + outbound.source_envelope #>> + '{callback_reply,execution_kind}', + outbound.source_envelope ->> 'execution_kind' + ) = 'no_write_replay' + AND coalesce( + outbound.source_envelope + ->> 'automation_run_id', + outbound.source_envelope #>> + '{callback_reply,automation_run_id}', + outbound.source_envelope #>> + '{source_refs,automation_run_ids,0}' + ) = :automation_run_id + AND coalesce( + outbound.source_envelope #>> + '{callback_reply,incident_id}', + outbound.source_envelope #>> + '{source_refs,incident_ids,0}' + ) = :incident_id + AND outbound.source_envelope #>> + '{callback_reply,apply_op_id}' + = :terminal_op_id + ) + """), + { + "project_id": project_id, + "automation_run_id": automation_run_id, + "incident_id": claim.incident_id, + "terminal_op_id": terminal_op_id, + }, + ) + return result.scalar() is True + except Exception as exc: + logger.warning( + "ansible_stdin_boundary_telegram_readback_failed", + automation_run_id=automation_run_id, + terminal_op_id=terminal_op_id, + error_type=type(exc).__name__, + ) + return False + + +async def _record_stdin_boundary_incident_terminal_disposition( + claim: AnsibleCheckModeClaim, + *, + terminal_op_id: str, + learning_stage_ids: list[str], + project_id: str, +) -> dict[str, Any] | None: + automation_run_id = _automation_run_id_for_claim(claim) + terminal_payload = { + **_incident_terminal_disposition_payload( + claim, + apply_op_id=terminal_op_id, + terminal_type="no_write_stdin_boundary_replay_terminal", + success_terminal=False, + retry_op_id=claim.op_id, + telegram_receipt_acknowledged=True, + ), + "schema_version": ( + "ansible_stdin_boundary_incident_terminal_v1" + ), + "terminal_op_id": terminal_op_id, + "proposal_executed": False, + "execution_success": None, + "learning_stage_ids": learning_stage_ids, + "learning_writeback_acknowledged": True, + "repository_readback_verified": True, + "runtime_apply_executed": False, + } + try: + async with get_db_context(project_id) as db: + selected = await db.execute( + text(""" + SELECT status::text AS incident_status, outcome, updated_at + FROM incidents + WHERE incident_id = :incident_id + AND project_id = :project_id + LIMIT 1 + FOR UPDATE + """), + { + "incident_id": claim.incident_id, + "project_id": project_id, + }, + ) + current = selected.mappings().one_or_none() + if current is None: + return None + current_status = str(current.get("incident_status") or "") + projection_superseded = current_status.upper() in { + "RESOLVED", + "CLOSED", + } + incident_write_performed = False + readback = current + if not projection_superseded: + updated = await db.execute( + text(""" + UPDATE incidents + SET status = 'MITIGATING'::incidentstatus, + outcome = CAST( + coalesce( + CASE + WHEN outcome IS NULL THEN '{}'::jsonb + WHEN jsonb_typeof(CAST(outcome AS jsonb)) + = 'object' + THEN CAST(outcome AS jsonb) + ELSE jsonb_build_object( + 'legacy_outcome', + CAST(outcome AS jsonb) + ) + END, + '{}'::jsonb + ) || jsonb_build_object( + 'automation_terminal', + CAST(:terminal_payload AS jsonb), + 'proposal_executed', + false, + 'execution_success', + NULL + ) AS json + ), + updated_at = NOW() + WHERE incident_id = :incident_id + AND project_id = :project_id + AND upper(status::text) + NOT IN ('RESOLVED', 'CLOSED') + RETURNING + status::text AS incident_status, + outcome, + updated_at + """), + { + "terminal_payload": json.dumps( + terminal_payload, + ensure_ascii=False, + ), + "incident_id": claim.incident_id, + "project_id": project_id, + }, + ) + readback = updated.mappings().one_or_none() + incident_write_performed = readback is not None + if readback is None: + return None + outcome = _json_loads(readback.get("outcome")) + terminal_readback = outcome.get("automation_terminal") + if not ( + isinstance(terminal_readback, Mapping) + and terminal_readback.get("automation_run_id") + == automation_run_id + and terminal_readback.get("terminal_op_id") + == terminal_op_id + and terminal_readback.get("runtime_apply_executed") + is False + ): + return None + return { + **terminal_payload, + "incident_status": str( + readback.get("incident_status") or current_status + ), + "incident_projection_write_performed": ( + incident_write_performed + ), + "incident_projection_superseded": projection_superseded, + "incident_projection_readback_verified": True, + "incident_row_version": ( + readback["updated_at"].isoformat() + if readback.get("updated_at") is not None + else None + ), + } + except Exception as exc: + logger.warning( + "ansible_stdin_boundary_incident_projection_failed", + automation_run_id=automation_run_id, + terminal_op_id=terminal_op_id, + error_type=type(exc).__name__, + ) + return None + + +async def _verify_stdin_boundary_terminal_projection_readback( + claim: AnsibleCheckModeClaim, + *, + terminal_op_id: str, + project_id: str, + require_projection_flag: bool, +) -> bool: + automation_run_id = _automation_run_id_for_claim(claim) + try: + async with get_db_context(project_id) as db: + await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) + result = await db.execute( + text(""" + SELECT EXISTS ( + SELECT 1 + FROM automation_operation_log terminal + JOIN automation_operation_log replay + ON replay.op_id = terminal.parent_op_id + AND replay.operation_type + = 'ansible_check_mode_executed' + AND replay.input ->> 'execution_mode' + = 'stdin_boundary_check_mode_replay' + JOIN automation_operation_log failed_check + ON failed_check.op_id::text + = replay.input + ->> 'replay_of_check_mode_op_id' + AND failed_check.operation_type + = 'ansible_check_mode_executed' + AND failed_check.status = 'failed' + AND failed_check.parent_op_id = replay.parent_op_id + AND failed_check.input ->> 'automation_run_id' + = :automation_run_id + JOIN automation_operation_log candidate + ON candidate.op_id = replay.parent_op_id + AND candidate.operation_type + = 'ansible_candidate_matched' + AND coalesce( + nullif( + candidate.input + ->> 'automation_run_id', + '' + ), + candidate.op_id::text + ) = :automation_run_id + WHERE terminal.op_id + = CAST(:terminal_op_id AS uuid) + AND terminal.operation_type + = 'ansible_execution_skipped' + AND terminal.status IN ('success', 'failed') + AND terminal.input ->> 'execution_mode' + = 'stdin_boundary_check_mode_replay_terminal' + AND terminal.input ->> 'automation_run_id' + = :automation_run_id + AND terminal.input + ->> 'retry_of_check_mode_op_id' + = failed_check.op_id::text + AND replay.input ->> 'automation_run_id' + = :automation_run_id + AND coalesce( + terminal.incident_id::text, + terminal.input ->> 'incident_id' + ) = :incident_id + AND terminal.output ->> 'runtime_apply_executed' + = 'false' + AND ( + SELECT COUNT(DISTINCT receipt.value + ->> 'stage_id') + FROM jsonb_array_elements( + coalesce( + terminal.input + -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) receipt(value) + WHERE receipt.value ->> 'stage_id' IN ( + 'mcp_context', + 'service_log_evidence', + 'normalized_asset_identity', + 'source_truth_diff', + 'risk_policy_decision', + 'executor_log_projection', + 'retry_or_rollback', + 'km_playbook_writeback', + 'rag_writeback', + 'playbook_trust', + 'timeline_projection' + ) + AND receipt.value ->> 'automation_run_id' + = :automation_run_id + AND receipt.value ->> 'durable_receipt' + = 'true' + ) = 11 + AND EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + terminal.input + -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) retry_receipt(value) + WHERE retry_receipt.value ->> 'stage_id' + = 'retry_or_rollback' + AND retry_receipt.value #>> + '{detail,verified_no_write_terminal}' + = 'true' + AND retry_receipt.value #>> + '{detail,runtime_apply_executed}' + = 'false' + AND retry_receipt.value #>> + '{detail,repository_readback_verified}' + = 'true' + AND retry_receipt.value + ->> 'durable_receipt' = 'true' + ) + AND EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + terminal.input + -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) closure_receipt(value) + WHERE closure_receipt.value ->> 'stage_id' + = 'incident_closure' + AND closure_receipt.value + ->> 'automation_run_id' + = :automation_run_id + AND closure_receipt.value #>> + '{detail,terminal_op_id}' + = CAST(:terminal_op_id AS text) + AND closure_receipt.value #>> + '{detail,no_write_terminal}' = 'true' + AND closure_receipt.value #>> + '{detail,telegram_receipt_acknowledged}' + = 'true' + AND closure_receipt.value #>> + '{detail,learning_writeback_acknowledged}' + = 'true' + AND closure_receipt.value #>> + '{detail,incident_projection_readback_verified}' + = 'true' + AND closure_receipt.value #>> + '{detail,repository_readback_verified}' + = 'true' + AND ( + :require_projection_flag IS FALSE + OR closure_receipt.value #>> + '{detail,projection_readback_verified}' + = 'true' + ) + AND closure_receipt.value + ->> 'durable_receipt' = 'true' + ) + AND EXISTS ( + SELECT 1 + FROM awooop_outbound_message outbound + WHERE outbound.project_id = :project_id + AND outbound.channel_type = 'telegram' + AND outbound.send_status = 'sent' + AND outbound.provider_message_id IS NOT NULL + AND outbound.source_envelope #>> + '{callback_reply,action}' + = 'controlled_apply_result' + AND outbound.source_envelope #>> + '{callback_reply,execution_kind}' + = 'no_write_replay' + AND outbound.source_envelope #>> + '{callback_reply,automation_run_id}' + = :automation_run_id + AND outbound.source_envelope #>> + '{callback_reply,incident_id}' + = :incident_id + AND outbound.source_envelope #>> + '{callback_reply,apply_op_id}' + = CAST(:terminal_op_id AS text) + ) + AND EXISTS ( + SELECT 1 + FROM alert_operation_log lifecycle + WHERE lifecycle.incident_id = :incident_id + AND lifecycle.event_type::text + = 'EXECUTION_COMPLETED' + AND lifecycle.success IS FALSE + AND lifecycle.context + ->> 'automation_run_id' + = :automation_run_id + AND lifecycle.context ->> 'apply_op_id' + = CAST(:terminal_op_id AS text) + ) + AND EXISTS ( + SELECT 1 + FROM alert_operation_log lifecycle + WHERE lifecycle.incident_id = :incident_id + AND lifecycle.event_type::text + = 'TELEGRAM_RESULT_SENT' + AND lifecycle.context + ->> 'automation_run_id' + = :automation_run_id + AND lifecycle.context ->> 'apply_op_id' + = CAST(:terminal_op_id AS text) + ) + ) + """), + { + "terminal_op_id": terminal_op_id, + "automation_run_id": automation_run_id, + "incident_id": claim.incident_id, + "project_id": project_id, + "require_projection_flag": require_projection_flag, + }, + ) + return result.scalar() is True + except Exception as exc: + logger.warning( + "ansible_stdin_boundary_projection_readback_failed", + automation_run_id=automation_run_id, + terminal_op_id=terminal_op_id, + error_type=type(exc).__name__, + ) + return False + + +async def backfill_missing_stdin_boundary_terminal_projections_once( + *, + project_id: str = "awoooi", + window_hours: int = _STDIN_BOUNDARY_REPLAY_WINDOW_HOURS, + limit: int = 1, +) -> dict[str, Any]: + """Close no-write replay attempts without re-executing their apply.""" + + stats: dict[str, Any] = { + "scanned": 0, + "written": 0, + "learning_written": 0, + "telegram_receipt_acknowledged": 0, + "incident_receipt_written": 0, + "lifecycle_written": 0, + "verified": 0, + "runtime_apply_executed": False, + "blockers": [], + "error": None, + } + try: + rows = await _load_missing_stdin_boundary_terminal_projection_rows( + project_id=project_id, + window_hours=window_hours, + limit=limit, + ) + stats["scanned"] = len(rows) + for row in rows: + claim = _claim_from_stdin_boundary_terminal_row(row) + terminal_op_id = str(row.get("terminal_op_id") or "") + terminal_output = _json_loads(row.get("terminal_output")) + if claim is None or not terminal_op_id: + stats["blockers"].append("terminal_claim_reconstruction_failed") + continue + returncode = _int_from_value( + terminal_output.get("check_mode_replay_returncode"), + default=( + 0 + if str(row.get("terminal_status") or "") == "success" + else 1 + ), + ) + replay_result = AnsibleRunResult( + returncode=returncode, + stdout="", + stderr=str(row.get("terminal_error") or ""), + duration_ms=_int_from_value( + row.get("terminal_duration_ms"), + default=0, + ), + post_verifier_passed=False, + ) + learning = await _record_no_write_replay_learning_receipts( + claim, + replay_result, + terminal_op_id=terminal_op_id, + project_id=project_id, + ) + if learning.get("ready") is not True: + missing = ",".join(learning.get("missing_stage_ids") or []) + stats["blockers"].append( + f"learning_receipts_missing:{missing}"[:300] + ) + continue + learning_written = ( + await _append_runtime_stage_receipts_to_operation( + operation_id=terminal_op_id, + operation_type="ansible_execution_skipped", + receipts=tuple(learning.get("receipts") or ()), + project_id=project_id, + ) + ) + stats["learning_written"] += int(learning_written) + if not learning_written: + stats["blockers"].append("learning_receipt_append_failed") + continue + + telegram_acknowledged = ( + await _stdin_boundary_replay_telegram_receipt_acknowledged( + claim, + terminal_op_id=terminal_op_id, + project_id=project_id, + ) + ) + if not telegram_acknowledged: + receipt_sent = await _send_controlled_apply_telegram_receipt( + claim, + replay_result, + apply_op_id=terminal_op_id, + writeback={ + "verification_result": "failed", + "verification": True, + "learning": True, + }, + project_id=project_id, + execution_kind="no_write_replay", + ) + if receipt_sent: + telegram_acknowledged = ( + await _stdin_boundary_replay_telegram_receipt_acknowledged( + claim, + terminal_op_id=terminal_op_id, + project_id=project_id, + ) + ) + if not telegram_acknowledged: + stats["blockers"].append("telegram_receipt_pending") + continue + stats["telegram_receipt_acknowledged"] += 1 + + execution_lifecycle = await _append_alert_lifecycle_receipt( + claim, + "EXECUTION_COMPLETED", + apply_op_id=terminal_op_id, + success=False, + action_detail=( + "stdin_boundary_replay_verified_no_write_terminal" + ), + project_id=project_id, + error_message="ai_transport_or_playbook_repair_required", + post_verifier_passed=False, + ) + telegram_lifecycle = await _append_alert_lifecycle_receipt( + claim, + "TELEGRAM_RESULT_SENT", + apply_op_id=terminal_op_id, + success=False, + action_detail=( + "stdin_boundary_no_write_result_receipt_sent" + ), + project_id=project_id, + post_verifier_passed=False, + ) + if not execution_lifecycle or not telegram_lifecycle: + stats["blockers"].append("lifecycle_projection_pending") + continue + stats["lifecycle_written"] += 1 + + incident_detail = ( + await _record_stdin_boundary_incident_terminal_disposition( + claim, + terminal_op_id=terminal_op_id, + learning_stage_ids=list(learning["stage_ids"]), + project_id=project_id, + ) + ) + if incident_detail is None: + stats["blockers"].append("incident_projection_pending") + continue + closure_detail = { + **incident_detail, + "terminal_op_id": terminal_op_id, + "retry_check_mode_op_id": claim.op_id, + "check_mode_replay_returncode": replay_result.returncode, + "runtime_apply_executed": False, + "verified_no_write_terminal": True, + "telegram_receipt_acknowledged": True, + "learning_writeback_acknowledged": True, + "required_learning_stage_ids": sorted( + _NO_WRITE_REPLAY_REQUIRED_STAGE_IDS + ), + "projection_readback_verified": False, + "repository_readback_verified": True, + "safe_next_action": ( + "queue_ai_transport_or_playbook_repair_candidate" + ), + } + closure_receipt = _runtime_stage_receipt( + claim, + stage_id="incident_closure", + evidence_ref=( + f"automation_operation_log:{terminal_op_id}:" + "no_write_terminal" + ), + detail=closure_detail, + derived_from_durable_chain=True, + ) + incident_receipt_written = ( + await _append_runtime_stage_receipts_to_operation( + operation_id=terminal_op_id, + operation_type="ansible_execution_skipped", + receipts=(closure_receipt,), + project_id=project_id, + ) + ) + stats["incident_receipt_written"] += int( + incident_receipt_written + ) + if not incident_receipt_written: + stats["blockers"].append("incident_receipt_append_failed") + continue + verified = await _verify_stdin_boundary_terminal_projection_readback( + claim, + terminal_op_id=terminal_op_id, + project_id=project_id, + require_projection_flag=False, + ) + if not verified: + stats["blockers"].append("projection_readback_pending") + continue + final_receipt = _runtime_stage_receipt( + claim, + stage_id="incident_closure", + evidence_ref=( + f"automation_operation_log:{terminal_op_id}:" + "no_write_terminal" + ), + detail={ + **closure_detail, + "projection_readback_verified": True, + }, + derived_from_durable_chain=True, + ) + finalized = await _append_runtime_stage_receipts_to_operation( + operation_id=terminal_op_id, + operation_type="ansible_execution_skipped", + receipts=(final_receipt,), + project_id=project_id, + ) + if not finalized: + stats["blockers"].append("projection_finalization_pending") + continue + final_verified = ( + await _verify_stdin_boundary_terminal_projection_readback( + claim, + terminal_op_id=terminal_op_id, + project_id=project_id, + require_projection_flag=True, + ) + ) + stats["verified"] += int(final_verified) + stats["written"] += int(final_verified) + if not final_verified: + stats["blockers"].append("final_projection_readback_pending") + except Exception as exc: + stats["error"] = f"{type(exc).__name__}: {exc}"[:500] + logger.warning( + "ansible_stdin_boundary_projection_backfill_failed", + project_id=project_id, + **stats, + ) + stats["blockers"] = sorted(set(stats["blockers"])) + return stats + + async def _record_learning_writeback_receipt( claim: AnsibleCheckModeClaim, result: AnsibleRunResult, @@ -4032,6 +5260,7 @@ async def _send_controlled_apply_telegram_receipt( apply_op_id: str, writeback: dict[str, Any], project_id: str, + execution_kind: str = "controlled_apply", ) -> bool: try: from src.services.telegram_gateway import get_telegram_gateway @@ -4052,6 +5281,7 @@ async def _send_controlled_apply_telegram_receipt( duration_ms=result.duration_ms, verifier_written=bool(writeback.get("verification")), learning_written=bool(writeback.get("learning")), + execution_kind=execution_kind, project_id=project_id, ) return bool( @@ -6710,6 +7940,51 @@ async def run_pending_check_modes_once( effective_timeout_seconds = ( timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS ) + stdin_projection_stats = ( + await backfill_missing_stdin_boundary_terminal_projections_once( + project_id=project_id, + window_hours=_STDIN_BOUNDARY_REPLAY_WINDOW_HOURS, + limit=max(1, min(int(limit), 5)), + ) + ) + stdin_projection_summary = { + "stdin_terminal_projection_scanned": int( + stdin_projection_stats.get("scanned") or 0 + ), + "stdin_terminal_projection_written": int( + stdin_projection_stats.get("written") or 0 + ), + "stdin_terminal_projection_learning_written": int( + stdin_projection_stats.get("learning_written") or 0 + ), + "stdin_terminal_projection_telegram_acknowledged": int( + stdin_projection_stats.get("telegram_receipt_acknowledged") or 0 + ), + "stdin_terminal_projection_incident_receipt_written": int( + stdin_projection_stats.get("incident_receipt_written") or 0 + ), + "stdin_terminal_projection_verified": int( + stdin_projection_stats.get("verified") or 0 + ), + "stdin_terminal_projection_blockers": [ + str(item)[:300] + for item in stdin_projection_stats.get("blockers") or [] + ], + "stdin_terminal_projection_error": ( + str(stdin_projection_stats.get("error") or "")[:500] or None + ), + "stdin_terminal_projection_runtime_apply_executed": False, + } + if ( + stdin_projection_summary["stdin_terminal_projection_scanned"] + or stdin_projection_summary["stdin_terminal_projection_blockers"] + or stdin_projection_summary["stdin_terminal_projection_error"] + ): + logger.info( + "ansible_stdin_boundary_projection_priority_tick", + project_id=project_id, + **stdin_projection_summary, + ) receipt_stats = await backfill_missing_auto_repair_execution_receipts_once( project_id=project_id, window_hours=24, @@ -6824,6 +8099,7 @@ async def run_pending_check_modes_once( "failed": 0, "capability_expired": expired_capability_count, "blockers": blockers, + **stdin_projection_summary, **receipt_backfill_summary, **failed_apply_retry_summary, } @@ -6836,6 +8112,7 @@ async def run_pending_check_modes_once( "failed": 0, "capability_expired": expired_capability_count, "blockers": transport_blockers, + **stdin_projection_summary, **receipt_backfill_summary, **failed_apply_retry_summary, } @@ -7061,6 +8338,7 @@ async def run_pending_check_modes_once( "capability_expired": expired_capability_count, "capability_revoke_failed": capability_revoke_failed, "catalog_replay_error": catalog_replay_error, + **stdin_projection_summary, **receipt_backfill_summary, **failed_apply_retry_summary, "blockers": [], diff --git a/apps/api/src/services/awooop_ansible_learning_writeback.py b/apps/api/src/services/awooop_ansible_learning_writeback.py index c712f54d4..2e529d048 100644 --- a/apps/api/src/services/awooop_ansible_learning_writeback.py +++ b/apps/api/src/services/awooop_ansible_learning_writeback.py @@ -104,9 +104,15 @@ def _playbook_trust_readback( playbook: PlaybookRecord, identity: dict[str, Any], verification_result: str, + trust_mutation_performed: bool = True, + observation_kind: str = "controlled_apply", ) -> dict[str, Any]: return { - "schema_version": "ansible_playbook_trust_writeback_v2", + "schema_version": ( + "ansible_playbook_trust_writeback_v2" + if trust_mutation_performed + else "ansible_playbook_no_write_observation_v1" + ), "identity_schema_version": identity["schema_version"], "identity_fingerprint": identity["identity_fingerprint"], "canonical_playbook_id": playbook.playbook_id, @@ -125,8 +131,11 @@ def _playbook_trust_readback( ), "trust_score": round(float(playbook.trust_score or 0.0), 6), "verification_result": verification_result, + "observation_kind": observation_kind, "learning_recorded": True, - "trust_updated": True, + "trust_observation_recorded": True, + "trust_updated": trust_mutation_performed, + "trust_mutation_performed": trust_mutation_performed, "repository_write_acknowledged": True, "repository_readback_verified": True, "operation_receipt_readback_verified": True, @@ -146,10 +155,17 @@ def _existing_trust_readback_matches( *, playbook: PlaybookRecord, identity: dict[str, Any], + trust_mutation_performed: bool = True, + observation_kind: str = "controlled_apply", ) -> bool: + expected_schema_version = ( + "ansible_playbook_trust_writeback_v2" + if trust_mutation_performed + else "ansible_playbook_no_write_observation_v1" + ) return bool( existing_trust.get("schema_version") - == "ansible_playbook_trust_writeback_v2" + == expected_schema_version and existing_trust.get("canonical_playbook_id") == identity["canonical_playbook_id"] and existing_trust.get("identity_fingerprint") @@ -161,7 +177,22 @@ def _existing_trust_readback_matches( and existing_trust.get("repository_readback_verified") is True and existing_trust.get("operation_receipt_readback_verified") is True and existing_trust.get("learning_recorded") is True - and existing_trust.get("trust_updated") is True + and existing_trust.get("trust_updated") is trust_mutation_performed + and ( + existing_trust.get("trust_mutation_performed") + is trust_mutation_performed + or ( + trust_mutation_performed + and existing_trust.get("trust_mutation_performed") is None + ) + ) + and ( + existing_trust.get("observation_kind") == observation_kind + or ( + observation_kind == "controlled_apply" + and existing_trust.get("observation_kind") is None + ) + ) ) @@ -175,8 +206,10 @@ async def record_ansible_playbook_trust_writeback( playbook_path: str, apply_op_id: str, verification_result: str, + trust_mutation_performed: bool = True, + observation_kind: str = "controlled_apply", ) -> dict[str, Any] | None: - """Atomically update canonical PlayBook trust and its learning receipt.""" + """Persist a PlayBook trust mutation or a no-write observation receipt.""" catalog = get_ansible_catalog_item(catalog_id) identity = resolve_ansible_playbook_identity(catalog_id) @@ -237,6 +270,8 @@ async def record_ansible_playbook_trust_writeback( existing_trust, playbook=playbook, identity=identity, + trust_mutation_performed=trust_mutation_performed, + observation_kind=observation_kind, ): return existing_trust logger.warning( @@ -309,26 +344,29 @@ async def record_ansible_playbook_trust_writeback( db.add(playbook) await db.flush() - source_incident_ids = list(playbook.source_incident_ids or []) - if incident_id and incident_id not in source_incident_ids: - source_incident_ids.append(incident_id) - playbook.source_incident_ids = source_incident_ids + if trust_mutation_performed: + source_incident_ids = list(playbook.source_incident_ids or []) + if incident_id and incident_id not in source_incident_ids: + source_incident_ids.append(incident_id) + playbook.source_incident_ids = source_incident_ids - if success: - playbook.success_count = int(playbook.success_count or 0) + 1 - playbook.trust_score = 0.9 * float(playbook.trust_score) + 0.1 - else: - playbook.failure_count = int(playbook.failure_count or 0) + 1 - playbook.trust_score = 0.8 * float(playbook.trust_score) - playbook.trust_score = max(0.0, min(1.0, playbook.trust_score)) - playbook.last_used_at = now - playbook.updated_at = now + if success: + playbook.success_count = int(playbook.success_count or 0) + 1 + playbook.trust_score = 0.9 * float(playbook.trust_score) + 0.1 + else: + playbook.failure_count = int(playbook.failure_count or 0) + 1 + playbook.trust_score = 0.8 * float(playbook.trust_score) + playbook.trust_score = max(0.0, min(1.0, playbook.trust_score)) + playbook.last_used_at = now + playbook.updated_at = now await db.flush() await db.refresh(playbook) trust_writeback = _playbook_trust_readback( playbook=playbook, identity=identity, verification_result=verification_result, + trust_mutation_performed=trust_mutation_performed, + observation_kind=observation_kind, ) input_payload = { "schema_version": "ansible_learning_writeback_receipt_v3", @@ -341,6 +379,8 @@ async def record_ansible_playbook_trust_writeback( "playbook_path": playbook_path, "apply_op_id": apply_op_id, "verification_result": verification_result, + "observation_kind": observation_kind, + "trust_mutation_performed": trust_mutation_performed, "learning_repository": "playbooks", "writer_source_sha": trust_writeback["writer_source_sha"], "stores_raw_logs": False, @@ -348,7 +388,9 @@ async def record_ansible_playbook_trust_writeback( } output_payload = { "learning_recorded": True, - "trust_updated": True, + "trust_observation_recorded": True, + "trust_updated": trust_mutation_performed, + "trust_mutation_performed": trust_mutation_performed, "success": success, "repository_readback_verified": True, "playbook_trust_writeback": trust_writeback, @@ -360,7 +402,7 @@ async def record_ansible_playbook_trust_writeback( input, output, dry_run_result, parent_op_id, tags ) VALUES ( 'ansible_learning_writeback_recorded', - 'ansible_controlled_apply_worker', + :actor, 'success', :incident_db_id, CAST(:input AS jsonb), @@ -375,9 +417,14 @@ async def record_ansible_playbook_trust_writeback( "input": json.dumps(input_payload, ensure_ascii=False), "output": json.dumps(output_payload, ensure_ascii=False), "parent_op_id": apply_op_id, + "actor": ( + "ansible_controlled_apply_worker" + if trust_mutation_performed + else "ansible_no_write_replay_worker" + ), "tags": [ "ansible", - "controlled_apply", + observation_kind, "learning_writeback", "playbook_trust", "ai_agent_auto_execution", @@ -409,7 +456,11 @@ async def record_ansible_playbook_trust_writeback( operation_row and operation_row.get("status") == "success" and operation_output.get("learning_recorded") is True - and operation_output.get("trust_updated") is True + and operation_output.get("trust_observation_recorded") is True + and operation_output.get("trust_updated") + is trust_mutation_performed + and operation_output.get("trust_mutation_performed") + is trust_mutation_performed and operation_output.get("repository_readback_verified") is True and operation_trust.get("playbook_row_fingerprint") @@ -435,10 +486,11 @@ async def ensure_ansible_rag_writeback( project_id: str, incident_id: str, apply_op_id: str, + path_type: str | None = None, ) -> dict[str, Any] | None: """Persist and read back the KM embedding used by semantic retrieval.""" - path_type = f"ansible_apply_receipt:{apply_op_id[:8]}" + effective_path_type = path_type or f"ansible_apply_receipt:{apply_op_id[:8]}" try: async with get_db_context(project_id) as db: selected = await db.execute( @@ -453,7 +505,10 @@ async def ensure_ansible_rag_writeback( AND path_type = :path_type LIMIT 1 """), - {"incident_id": incident_id, "path_type": path_type}, + { + "incident_id": incident_id, + "path_type": effective_path_type, + }, ) row = selected.mappings().first() if not row: 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 659981f14..10f21e35e 100644 --- a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py +++ b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py @@ -88,9 +88,24 @@ _RUNTIME_LIFECYCLE_SQL = """ ) ELSE NULL END - AND apply.operation_type - = 'ansible_apply_executed' - AND apply.status = 'failed' + AND ( + ( + apply.operation_type + = 'ansible_apply_executed' + AND apply.status = 'failed' + ) + OR ( + apply.operation_type + = 'ansible_execution_skipped' + AND apply.status + IN ('success', 'failed') + AND apply.input ->> 'execution_mode' + = 'stdin_boundary_check_mode_replay_terminal' + AND apply.output #>> + '{runtime_apply_executed}' + = 'false' + ) + ) AND apply.input ->> 'automation_run_id' = COALESCE( NULLIF( @@ -148,7 +163,59 @@ _RUNTIME_LIFECYCLE_SQL = """ AND closure_receipt.value #>> '{detail,repository_readback_verified}' = 'true' + AND ( + apply.operation_type + <> 'ansible_execution_skipped' + OR ( + closure_receipt.value #>> + '{detail,learning_writeback_acknowledged}' + = 'true' + AND closure_receipt.value #>> + '{detail,incident_projection_readback_verified}' + = 'true' + AND closure_receipt.value #>> + '{detail,projection_readback_verified}' + = 'true' + ) + ) ) + AND ( + apply.operation_type + <> 'ansible_execution_skipped' + OR ( + SELECT COUNT(DISTINCT + stage_receipt.value + ->> 'stage_id') + FROM jsonb_array_elements( + COALESCE( + apply.input + -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) stage_receipt(value) + WHERE stage_receipt.value + ->> 'stage_id' IN ( + 'mcp_context', + 'service_log_evidence', + 'normalized_asset_identity', + 'source_truth_diff', + 'risk_policy_decision', + 'executor_log_projection', + 'retry_or_rollback', + 'km_playbook_writeback', + 'rag_writeback', + 'playbook_trust', + 'timeline_projection' + ) + AND stage_receipt.value + ->> 'automation_run_id' + = apply.input + ->> 'automation_run_id' + AND stage_receipt.value + ->> 'durable_receipt' + = 'true' + ) = 11 + ) ) AND EXISTS ( SELECT 1 diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 246be013e..855a068b8 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -10086,6 +10086,7 @@ class TelegramGateway: duration_ms: int, verifier_written: bool, learning_written: bool, + execution_kind: str = "controlled_apply", project_id: str = "awoooi", ) -> dict: """Send and mirror the AI Agent controlled-apply result receipt.""" @@ -10098,16 +10099,29 @@ class TelegramGateway: ) return {} - success = verification_result == "success" and returncode == 0 + no_write_replay = execution_kind == "no_write_replay" + success = ( + not no_write_replay + and verification_result == "success" + and returncode == 0 + ) next_step = ( "monitor_for_regression" if success - else "queue_ai_rollback_or_playbook_repair" + else ( + "queue_ai_transport_or_playbook_repair" + if no_write_replay + else "queue_ai_rollback_or_playbook_repair" + ) ) title = ( "CONTROLLED APPLY RESULT｜AI Agent 受控執行完成" if success - else "CONTROLLED APPLY RESULT｜AI Agent 受控執行待修復" + else ( + "NO-WRITE RETRY RESULT｜AI Agent 無寫入重試待修復" + if no_write_replay + else "CONTROLLED APPLY RESULT｜AI Agent 受控執行待修復" + ) ) truth_chain: dict[str, object] | None = None try: @@ -10147,11 +10161,13 @@ class TelegramGateway: if source_extra is not None: source_extra["outbound_message_type"] = "final" if success else "error" source_extra["automation_run_id"] = automation_run_id + source_extra["execution_kind"] = execution_kind callback_reply = source_extra.get("callback_reply") if isinstance(callback_reply, dict): callback_reply["automation_run_id"] = automation_run_id callback_reply["apply_op_id"] = apply_op_id callback_reply["project_id"] = project_id or "awoooi" + callback_reply["execution_kind"] = execution_kind source_refs = source_extra.get("source_refs") if isinstance(source_refs, dict): source_refs["automation_run_ids"] = [automation_run_id] @@ -10160,7 +10176,10 @@ class TelegramGateway: f"Run: {html.escape(str(automation_run_id or '')[:8])}", f"Incident: {html.escape(str(incident_id))}", f"Catalog: {html.escape(str(catalog_id or '--'))}", - f"Apply op: {html.escape(str(apply_op_id or '')[:8])}", + ( + f"{'Terminal' if no_write_replay else 'Apply'} op: " + f"{html.escape(str(apply_op_id or '')[:8])}" + ), f"PlayBook: {html.escape(str(playbook_path or '--'))}", ( "Result: " @@ -10173,6 +10192,11 @@ class TelegramGateway: f"{html.escape(_bool_code(verifier_written))} / KM " f"{html.escape(_bool_code(learning_written))}" ), + ( + "Runtime apply: 0" + if no_write_replay + else "Runtime apply: 1" + ), f"Next: {html.escape(next_step)}", f"Runs: {html.escape(incident_runs_url(incident_id, project_id=project_id or 'awoooi'))}", ] 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 02d9156c8..e2baa3fc1 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -25,6 +25,52 @@ from src.services.ai_automation_runtime_contract import ( ) +def test_runtime_switches_distinguish_api_process_from_dedicated_broker() -> None: + payload = { + "runtime_switches": { + "ansible_check_mode_worker_enabled": False, + } + } + boundary = { + "status": "verified_ready", + "production_boundary_verified": True, + "source_sha_matches_deployment": True, + "receipt_ref": ( + "configmap:awoooi-prod/awoooi-executor-boundary-verification" + ), + "verified_source_sha": "a" * 40, + "controls": { + "broker_ssh_mount_present": True, + "broker_ssh_egress_allowlisted": True, + "broker_kubernetes_token_absent": True, + }, + } + + runtime_control_module._attach_effective_executor_runtime_switches( + payload, + boundary_readback=boundary, + ) + + switches = payload["runtime_switches"] + assert switches["ansible_check_mode_worker_enabled"] is False + assert switches["ansible_check_mode_worker_scope"] == "api_process_only" + assert switches["dedicated_execution_broker_ready"] is True + assert switches["effective_ansible_executor_available"] is True + assert switches["effective_executor_readback_source"] == ( + boundary["receipt_ref"] + ) + assert switches["effective_executor_verified_source_sha"] == "a" * 40 + + boundary["source_sha_matches_deployment"] = False + runtime_control_module._attach_effective_executor_runtime_switches( + payload, + boundary_readback=boundary, + ) + assert switches["dedicated_execution_broker_ready"] is False + assert switches["effective_ansible_executor_available"] is False + assert switches["effective_executor_readback_source"] is None + + def _closed_retry_terminal_row() -> dict[str, object]: run_id = "retry-run-1" apply_op_id = "94925d5e-7d73-4ed2-82e2-a3f2d3026270" diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index fe5eedd99..afc14222f 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -600,6 +600,12 @@ def test_terminal_candidate_contract_requires_verified_four_node_chain() -> None assert "apply.parent_op_id = check_mode.op_id" in predicate assert "replay.parent_op_id = apply.op_id" in predicate assert "replay.status IN ('success', 'failed')" in predicate + assert ( + "replay_check.parent_op_id = failed_check.parent_op_id" + in predicate + ) + assert "replay_of_check_mode_op_id" in predicate + assert "failed_check.parent_op_id = candidate.op_id" in predicate assert "verified_no_write_terminal" in predicate assert "repository_readback_verified" in predicate assert "{detail,execution_success}" in predicate diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 72a4b36f5..fbf23630e 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -25,6 +25,7 @@ from src.services.awooop_ansible_check_mode_service import ( AnsibleRunResult, _append_alert_lifecycle_receipt, _append_runtime_stage_receipts_to_apply, + _append_runtime_stage_receipts_to_operation, _automation_operation_log_incident_id, _build_auto_repair_execution_receipt, _build_retry_runtime_stage_receipt, @@ -33,21 +34,26 @@ from src.services.awooop_ansible_check_mode_service import ( _execution_capability_timeout_seconds, _expire_stale_ansible_execution_capabilities, _issue_ansible_execution_capability, + _load_missing_stdin_boundary_terminal_projection_rows, _load_open_failed_apply_retry_row, _load_pre_decision_context_runtime_stage_receipts, _post_apply_km_path_type, _post_apply_verification_result, _record_auto_repair_execution_receipt, _record_learning_writeback_receipt, + _record_no_write_replay_learning_receipts, _record_retry_runtime_stage_receipt, _record_runtime_stage_receipts, + _record_stdin_boundary_incident_terminal_disposition, _record_stdin_boundary_replay_terminal, _record_timeline_projection_receipt, _resolve_execution_capability_operation_type, _revoke_ansible_execution_capability, _run_ansible_command, _send_controlled_apply_telegram_receipt, + _verify_stdin_boundary_terminal_projection_readback, backfill_missing_auto_repair_execution_receipts_once, + backfill_missing_stdin_boundary_terminal_projections_once, build_ansible_apply_command, build_ansible_check_mode_claim_input, build_ansible_check_mode_command, @@ -159,6 +165,13 @@ def test_ansible_audit_keeps_external_incident_id_in_json_not_bigint_column() -> assert "candidate.input ->> 'incident_id'" in decision_source assert "verified_ansible_candidate_terminal_sql" in decision_source assert "ansible_execution_skipped" in terminal_source + assert "stdin_boundary_check_mode_replay_terminal" in terminal_source + assert "learning_writeback_acknowledged" in terminal_source + assert "incident_projection_readback_verified" in terminal_source + assert "projection_readback_verified" in terminal_source + assert "TELEGRAM_RESULT_SENT" in terminal_source + assert "no_write_replay" in terminal_source + assert "= 11" in terminal_source assert "operation_type, actor, status, incident_id" in claim_source assert "incident_db_id" in decision_source assert "incident_db_id" in claim_source @@ -2089,6 +2102,226 @@ def test_failed_stdin_boundary_checks_replay_once_without_apply() -> None: assert "stdin_boundary_replay_passed_no_write_terminal" in run_source +def test_stdin_boundary_terminal_projection_closes_only_from_durable_receipts( +) -> None: + backfill_source = inspect.getsource( + backfill_missing_stdin_boundary_terminal_projections_once + ) + learning_source = inspect.getsource( + _record_no_write_replay_learning_receipts + ) + incident_source = inspect.getsource( + _record_stdin_boundary_incident_terminal_disposition + ) + loader_source = inspect.getsource( + _load_missing_stdin_boundary_terminal_projection_rows + ) + verifier_source = inspect.getsource( + _verify_stdin_boundary_terminal_projection_readback + ) + + assert "run_controlled_apply_for_claim" not in backfill_source + assert 'execution_kind="no_write_replay"' in backfill_source + assert "runtime_apply_executed" in backfill_source + assert "projection_readback_verified" in backfill_source + assert "EntryType.ANTI_PATTERN" in learning_source + assert "EntryStatus.PUBLISHED" in learning_source + assert "ensure_ansible_rag_writeback" in learning_source + assert "record_ansible_playbook_trust_writeback" in learning_source + assert 'verification_result="not_applicable_no_runtime_apply"' in learning_source + assert "trust_mutation_performed=False" in learning_source + assert 'observation_kind="no_write_replay"' in learning_source + assert "mcp_context" in learning_source + assert "controlled_apply_executed" in learning_source + assert "historical_no_write_replay" in learning_source + assert "verified_no_write_terminal_pending_ai_repair" in learning_source + assert "incident_projection_superseded" in incident_source + for source in (loader_source, verifier_source): + assert "JOIN automation_operation_log failed_check" in source + assert "candidate.op_id = replay.parent_op_id" in source + assert "failed_check.parent_op_id = replay.parent_op_id" in source + assert "replay_of_check_mode_op_id" in source + assert "retry_of_check_mode_op_id" in source + assert "TELEGRAM_RESULT_SENT" in verifier_source + assert "km_playbook_writeback" in verifier_source + assert "playbook_trust" in verifier_source + assert "runtime_apply_executed" in verifier_source + assert verifier_source.count( + "CAST(:terminal_op_id AS text)" + ) == 4 + + +@pytest.mark.asyncio +async def test_stdin_boundary_terminal_projection_runs_no_write_pipeline( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.services import awooop_ansible_check_mode_service as service + + claim = AnsibleCheckModeClaim( + op_id="00000000-0000-0000-0000-000000000322", + source_candidate_op_id="00000000-0000-0000-0000-000000000321", + incident_id="INC-STDIN-PROJECTION", + catalog_id="ansible:188-momo-backup-user", + playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml", + apply_playbook_path=( + "infra/ansible/playbooks/188-momo-backup-user.yml" + ), + inventory_hosts=("host_188",), + risk_level="low", + input_payload={ + "automation_run_id": "00000000-0000-0000-0000-000000000321", + "historical_stdin_boundary_replay": True, + "controlled_apply_allowed": False, + "apply_enabled": False, + }, + ) + terminal_op_id = "00000000-0000-0000-0000-000000000323" + calls: list[tuple[str, object]] = [] + telegram_readbacks = iter((False, True)) + + async def load_rows(**_kwargs): + return [ + { + "terminal_op_id": terminal_op_id, + "terminal_status": "failed", + "terminal_output": { + "check_mode_replay_returncode": 2, + "runtime_apply_executed": False, + }, + "terminal_error": "stdin_boundary_repair_required", + "terminal_duration_ms": 71, + } + ] + + async def learning_receipts(*_args, **_kwargs): + calls.append(("learning", _kwargs["terminal_op_id"])) + return { + "ready": True, + "receipts": ({"stage_id": "mcp_context"},), + "stage_ids": sorted(service._NO_WRITE_REPLAY_REQUIRED_STAGE_IDS), + "missing_stage_ids": [], + } + + async def append_receipts(**kwargs): + calls.append( + ( + "append", + ( + kwargs["operation_id"], + kwargs["operation_type"], + tuple( + item["stage_id"] for item in kwargs["receipts"] + ), + ), + ) + ) + return True + + async def telegram_readback(*_args, **_kwargs): + calls.append(("telegram_readback", _kwargs["terminal_op_id"])) + return next(telegram_readbacks) + + async def send_telegram(*_args, **kwargs): + calls.append(("telegram_send", kwargs["execution_kind"])) + return True + + async def append_lifecycle(_claim, event_type, **kwargs): + calls.append( + ( + "lifecycle", + (event_type, kwargs["apply_op_id"], kwargs["success"]), + ) + ) + return True + + async def incident_projection(*_args, **_kwargs): + calls.append(("incident_projection", _kwargs["terminal_op_id"])) + return { + "no_write_terminal": True, + "incident_projection_readback_verified": True, + "repository_readback_verified": True, + } + + verifier_flags: list[bool] = [] + + async def verify_projection(*_args, **kwargs): + verifier_flags.append(kwargs["require_projection_flag"]) + return True + + monkeypatch.setattr( + service, + "_load_missing_stdin_boundary_terminal_projection_rows", + load_rows, + ) + monkeypatch.setattr( + service, + "_claim_from_stdin_boundary_terminal_row", + lambda _row: claim, + ) + monkeypatch.setattr( + service, + "_record_no_write_replay_learning_receipts", + learning_receipts, + ) + monkeypatch.setattr( + service, + "_append_runtime_stage_receipts_to_operation", + append_receipts, + ) + monkeypatch.setattr( + service, + "_stdin_boundary_replay_telegram_receipt_acknowledged", + telegram_readback, + ) + monkeypatch.setattr( + service, + "_send_controlled_apply_telegram_receipt", + send_telegram, + ) + monkeypatch.setattr( + service, + "_append_alert_lifecycle_receipt", + append_lifecycle, + ) + monkeypatch.setattr( + service, + "_record_stdin_boundary_incident_terminal_disposition", + incident_projection, + ) + monkeypatch.setattr( + service, + "_verify_stdin_boundary_terminal_projection_readback", + verify_projection, + ) + + result = await backfill_missing_stdin_boundary_terminal_projections_once( + project_id="awoooi", + limit=1, + ) + + assert result["scanned"] == 1 + assert result["written"] == 1 + assert result["verified"] == 1 + assert result["runtime_apply_executed"] is False + assert result["blockers"] == [] + assert ("telegram_send", "no_write_replay") in calls + assert ( + "lifecycle", + ("EXECUTION_COMPLETED", terminal_op_id, False), + ) in calls + assert ( + "lifecycle", + ("TELEGRAM_RESULT_SENT", terminal_op_id, False), + ) in calls + append_calls = [value for name, value in calls if name == "append"] + assert len(append_calls) == 3 + assert all(item[0] == terminal_op_id for item in append_calls) + assert all(item[1] == "ansible_execution_skipped" for item in append_calls) + assert append_calls[1][2] == ("incident_closure",) + assert append_calls[2][2] == ("incident_closure",) + assert verifier_flags == [False, True] + + @pytest.mark.asyncio async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims( monkeypatch: pytest.MonkeyPatch, @@ -2443,10 +2676,51 @@ def test_ansible_learning_writeback_uses_canonical_durable_service() -> None: assert '"operation_receipt_readback_verified": True' in receipt_source assert "operation_readback" in durable_source assert 'operation_output.get("learning_recorded") is True' in durable_source - assert 'operation_output.get("trust_updated") is True' in durable_source + assert "trust_mutation_performed: bool = True" in durable_source + assert 'operation_output.get("trust_updated")' in durable_source + assert "is trust_mutation_performed" in durable_source assert '"durable_write_acknowledged": True' in receipt_source +def test_no_write_playbook_observation_does_not_claim_trust_mutation() -> None: + playbook = SimpleNamespace( + playbook_id="PB-ANSIBLE-188-MOMO-BACKUP-USER", + project_id="awoooi", + version=3, + success_count=7, + failure_count=2, + trust_score=0.81, + updated_at=datetime(2026, 7, 14, tzinfo=UTC), + ) + identity = { + "schema_version": "ansible_playbook_identity_v1", + "identity_fingerprint": "a" * 64, + "canonical_playbook_id": playbook.playbook_id, + "catalog_id": "ansible:188-momo-backup-user", + "playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml", + } + + receipt = _playbook_trust_readback( + playbook=playbook, + identity=identity, + verification_result="not_applicable_no_runtime_apply", + trust_mutation_performed=False, + observation_kind="no_write_replay", + ) + + assert receipt["schema_version"] == ( + "ansible_playbook_no_write_observation_v1" + ) + assert receipt["observation_kind"] == "no_write_replay" + assert receipt["learning_recorded"] is True + assert receipt["trust_observation_recorded"] is True + assert receipt["trust_updated"] is False + assert receipt["trust_mutation_performed"] is False + assert receipt["success_count"] == 7 + assert receipt["failure_count"] == 2 + assert receipt["trust_score"] == 0.81 + + def test_ansible_catalog_ids_map_to_stable_playbook_primary_keys() -> None: expected = { "ansible:110-devops": "PB-ANSIBLE-110-DEVOPS", @@ -2492,6 +2766,9 @@ def test_ansible_rag_writeback_requires_embedding_readback() -> None: source = inspect.getsource(ensure_ansible_rag_writeback) fallback_source = inspect.getsource(_ensure_rag_chunk_fallback) + assert "path_type: str | None = None" in source + assert "path_type or f\"ansible_apply_receipt:" in source + assert '"path_type": effective_path_type' in source assert "ensure_entry_embedding" in source assert "embedding IS NOT NULL" in source assert "durable_write_acknowledged" in source @@ -2652,11 +2929,16 @@ def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None: receipt_builder_source = inspect.getsource( _build_retry_runtime_stage_receipt ) - append_source = inspect.getsource(_append_runtime_stage_receipts_to_apply) + append_source = inspect.getsource( + _append_runtime_stage_receipts_to_operation + ) assert "_build_retry_runtime_stage_receipt" in receipt_source assert "_runtime_stage_receipt" in receipt_builder_source assert 'stage_id="retry_or_rollback"' in receipt_builder_source assert "jsonb_array_elements" in append_source + assert "_append_runtime_stage_receipts_to_operation" in inspect.getsource( + _append_runtime_stage_receipts_to_apply + ) assert "ansible_controlled_retry_terminal_v2" in receipt_builder_source assert "ansible_retry_terminal_receipt_not_written_unverified" in receipt_source assert "runtime_apply_executed" in receipt_source diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py index 4356a0108..dc14e22fd 100644 --- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -61,6 +61,7 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None: for item in deployment["spec"]["template"]["spec"]["containers"][0]["env"] } assert worker_env["ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER"] == "true" + assert worker_env["AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS"] == "168" def test_execution_broker_is_only_workload_with_ssh_transport() -> None: 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 df562da36..533af7de7 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 @@ -805,13 +805,32 @@ def test_runtime_lifecycle_sql_accepts_only_durable_failed_terminals() -> None: assert "completed_terminal_count" in sql assert "apply.status = 'failed'" in sql + assert "stdin_boundary_check_mode_replay_terminal" in sql + assert "= 'ansible_execution_skipped'" 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,learning_writeback_acknowledged}'" in sql assert "'{detail,repository_readback_verified}'" in sql + assert "'{detail,incident_projection_readback_verified}'" in sql + assert "'{detail,projection_readback_verified}'" in sql + for stage_id in ( + "mcp_context", + "service_log_evidence", + "normalized_asset_identity", + "source_truth_diff", + "risk_policy_decision", + "executor_log_projection", + "retry_or_rollback", + "km_playbook_writeback", + "rag_writeback", + "playbook_trust", + "timeline_projection", + ): + assert stage_id in sql assert "telegram_result.event_type" in sql assert "= 'TELEGRAM_RESULT_SENT'" in sql assert "completed_success_count" not in sql diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py index 6b6a6fc4c..194af0064 100644 --- a/apps/api/tests/test_telegram_message_templates.py +++ b/apps/api/tests/test_telegram_message_templates.py @@ -1281,6 +1281,79 @@ async def test_controlled_apply_result_receipt_marks_callback_reply_evidence(mon assert snapshot["execution"]["ansible"]["controlled_apply"] is True +@pytest.mark.asyncio +async def test_no_write_replay_receipt_never_claims_runtime_apply(monkeypatch) -> None: + gateway = TelegramGateway() + sent_requests = [] + + async def fake_send_request(method, payload): + sent_requests.append((method, payload)) + return {"ok": True, "result": {"message_id": 12346}} + + async def fake_fetch_truth_chain(*, source_id, project_id): + assert source_id == "INC-NO-WRITE-001" + assert project_id == "awoooi" + return { + "truth_status": { + "current_stage": "execution_skipped", + "stage_status": "failed", + "needs_human": False, + "blockers": ["stdin_boundary_repair_required"], + }, + "automation_quality": { + "verdict": "ai_repair_queued", + "facts": {}, + "blockers": ["stdin_boundary_repair_required"], + }, + "execution": {"automation_operation_log": [], "ansible": {}}, + } + + async def fake_fetch_km_completion_summary(**_kwargs): + return {} + + monkeypatch.setattr(TelegramGateway, "alert_chat_id", property(lambda _self: "chat")) + monkeypatch.setattr(gateway, "_send_request", fake_send_request) + monkeypatch.setattr( + "src.services.awooop_truth_chain_service.fetch_truth_chain", + fake_fetch_truth_chain, + ) + monkeypatch.setattr( + "src.services.telegram_gateway._fetch_km_stale_completion_summary_for_incident", + fake_fetch_km_completion_summary, + ) + + result = await gateway.send_controlled_apply_result_receipt( + automation_run_id="00000000-0000-0000-0000-000000000002", + incident_id="INC-NO-WRITE-001", + catalog_id="ansible:188-momo-backup-user", + apply_op_id="83b7a95c-3652-4c0d-bb4c-729e500acedb", + playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml", + verification_result="success", + returncode=0, + duration_ms=721, + verifier_written=True, + learning_written=True, + execution_kind="no_write_replay", + project_id="awoooi", + ) + + assert result["ok"] is True + assert len(sent_requests) == 1 + method, payload = sent_requests[0] + assert method == "sendMessage" + assert "NO-WRITE RETRY RESULT" in payload["text"] + assert "AI Agent 無寫入重試待修復" in payload["text"] + assert "Terminal op:" in payload["text"] + assert "Runtime apply: 0" in payload["text"] + assert "queue_ai_transport_or_playbook_repair" in payload["text"] + assert "AI Agent 受控執行完成" not in payload["text"] + source_extra = payload["_awooop_source_envelope_extra"] + assert source_extra["execution_kind"] == "no_write_replay" + assert source_extra["outbound_message_type"] == "error" + assert source_extra["callback_reply"]["action"] == "controlled_apply_result" + assert source_extra["callback_reply"]["execution_kind"] == "no_write_replay" + + def test_callback_reply_awooop_status_chain_snapshot_marks_manual_gate() -> None: """Callback evidence 要保存當下 AwoooP 狀態鏈，不只保存 live query 結果。""" snapshot = telegram_gateway_module._callback_reply_awooop_status_chain_snapshot( diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index ae77f0c9d..7bf9869bf 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -183,7 +183,7 @@ spec: - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT value: "2" - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS - value: "24" + value: "168" - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS value: "10" - name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER