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 126c307c4..8251ed67a 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -162,6 +162,23 @@ _VERIFIED_CLOSURE_REQUIRED_STAGE_IDS = frozenset( "timeline_projection", } ) +_TELEGRAM_FINAL_RECEIPT_REQUIRED_PREREQUISITES = frozenset( + { + "candidate", + "check_mode", + "controlled_apply", + "post_apply_verifier", + "auto_repair_execution_receipt", + "km_playbook_writeback", + "rag_writeback", + "mcp_context", + "playbook_trust", + "timeline_projection", + "approval_projection", + "execution_lifecycle", + "required_runtime_stage_receipts", + } +) _NO_WRITE_REPLAY_REQUIRED_STAGE_IDS = _VERIFIED_CLOSURE_REQUIRED_STAGE_IDS _WAZUH_SCHEDULED_IDENTITY = { "iwooos_wazuh_manager_posture_scheduler": ( @@ -3161,6 +3178,15 @@ def _runtime_stage_ids(value: Any) -> set[str]: } +def _telegram_final_evidence_ready(receipts: Mapping[str, Any]) -> bool: + """Do not reserve the final outbox identity before evidence is complete.""" + + return all( + receipts.get(receipt_name) is True + for receipt_name in _TELEGRAM_FINAL_RECEIPT_REQUIRED_PREREQUISITES + ) + + async def _read_verified_apply_closure_prerequisites( claim: AnsibleCheckModeClaim, *, @@ -6684,6 +6710,7 @@ async def _send_controlled_apply_telegram_receipt( project_id: str, execution_kind: str = "controlled_apply", provider_delivery: str = "digest", + closure_receipts: Mapping[str, Any] | None = None, ) -> bool: try: from src.services.telegram_gateway import get_telegram_gateway @@ -6704,6 +6731,7 @@ async def _send_controlled_apply_telegram_receipt( ) if no_write_observation and verified_success: effective_provider_delivery = "shadow_only" + receipt_readback = dict(closure_receipts or {}) response = await get_telegram_gateway().send_controlled_apply_result_receipt( automation_run_id=str( claim.input_payload.get("automation_run_id") @@ -6719,7 +6747,22 @@ async def _send_controlled_apply_telegram_receipt( returncode=result.returncode, duration_ms=result.duration_ms, verifier_written=bool(writeback.get("verification")), - learning_written=bool(writeback.get("learning")), + learning_written=bool( + writeback.get("learning") + or receipt_readback.get("km_playbook_writeback") + ), + rag_written=bool( + writeback.get("rag_writeback") + or receipt_readback.get("rag_writeback") + ), + mcp_context_ready=bool(receipt_readback.get("mcp_context")), + playbook_trust_written=bool( + writeback.get("trust_learning") + or receipt_readback.get("playbook_trust") + ), + alert_category=str( + claim.input_payload.get("alert_category") or "" + ), execution_kind=effective_execution_kind, provider_delivery=effective_provider_delivery, project_id=project_id, @@ -6831,13 +6874,18 @@ async def _reconcile_verified_apply_closure_projections( telegram_receipt = bool( (readback.get("receipts") or {}).get("telegram_receipt") ) - if not telegram_receipt: + receipt_readback = readback.get("receipts") or {} + if ( + not telegram_receipt + and _telegram_final_evidence_ready(receipt_readback) + ): await _send_controlled_apply_telegram_receipt( claim, result, apply_op_id=apply_op_id, writeback=writeback, project_id=project_id, + closure_receipts=receipt_readback, ) readback = await _read_verified_apply_closure_prerequisites( claim, diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index b341242e7..13410e516 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -13399,6 +13399,10 @@ class TelegramGateway: duration_ms: int, verifier_written: bool, learning_written: bool, + rag_written: bool = False, + mcp_context_ready: bool = False, + playbook_trust_written: bool = False, + alert_category: str = "", execution_kind: str = "controlled_apply", provider_delivery: str = "digest", project_id: str = "awoooi", @@ -13412,6 +13416,36 @@ class TelegramGateway: and verification_result == "success" and returncode == 0 ) + evidence_coverage = { + "independent_verifier": bool(verifier_written), + "km_writeback": bool(learning_written), + "rag_writeback": bool(rag_written), + "mcp_context": bool(mcp_context_ready), + "playbook_trust": bool(playbook_trust_written), + } + evidence_complete = all(evidence_coverage.values()) + normalized_category = re.sub( + r"[^a-z0-9_.:-]+", + "_", + str(alert_category or "").strip().lower(), + )[:80].strip("_") + if not normalized_category: + normalized_category = ( + "alert_chain_health" + if catalog_id == "ansible:110-alertmanager-delivery-recovery" + else "windows_vmware" + if str(catalog_id).startswith("agent99:") + else "host_automation" + if str(catalog_id).startswith("ansible:") + else "controlled_automation" + ) + executor_agent = ( + "Agent99" + if str(catalog_id).startswith("agent99:") + else "Host Ansible executor" + if str(catalog_id).startswith("ansible:") + else "single controlled executor" + ) if no_write_posture: next_step = ( "record_healthy_posture_without_notification" @@ -13424,8 +13458,16 @@ class TelegramGateway: else "AI 自動修復摘要|姿態檢查待修復" ) elif success: - next_step = "monitor_for_regression" - title = "CONTROLLED APPLY RESULT|AI Agent 受控執行完成" + next_step = ( + "monitor_for_regression" + if evidence_complete + else "complete_missing_evidence_receipts" + ) + title = ( + "CONTROLLED APPLY RESULT|AI Agent 受控執行完成" + if evidence_complete + else "CONTROLLED APPLY RESULT|執行完成、證據待補" + ) elif no_write_replay: next_step = "queue_ai_transport_or_playbook_repair" title = "AI 自動修復摘要|PlayBook 檢查待修復" @@ -13479,7 +13521,13 @@ class TelegramGateway: awooop_status_chain=status_snapshot, ) if source_extra is not None: - source_extra["outbound_message_type"] = "final" if success else "error" + source_extra["outbound_message_type"] = ( + "final" + if success and (evidence_complete or no_write_posture) + else "interim" + 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") @@ -13492,6 +13540,14 @@ class TelegramGateway: callback_reply["runtime_write_performed"] = not ( no_write_replay or no_write_posture ) + callback_reply["alert_category"] = normalized_category + callback_reply["automation_evidence"] = { + "schema_version": "controlled_apply_automation_evidence_v1", + **evidence_coverage, + "complete": evidence_complete, + "executor_agent": executor_agent, + "llm_runtime_apply_authority": False, + } source_refs = source_extra.get("source_refs") if isinstance(source_refs, dict): source_refs["automation_run_ids"] = [automation_run_id] @@ -13618,25 +13674,48 @@ class TelegramGateway: else: lines = [ f"{html.escape(title)}", - 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"PlayBook: {html.escape(str(playbook_path or '--'))}", + "ai_automation_alert_card_v1", ( - "Result: " + "🏷 分類: " + f"{html.escape(normalized_category.upper())}" + ), + ( + "🎯 目標: " + f"{html.escape(str(catalog_id or '--'))}" + ), + ( + "🛠 執行: " f"{html.escape(str(verification_result or 'missing'))} " f"/ rc {html.escape(str(returncode))} " f"/ {html.escape(str(duration_ms))}ms" ), ( - "Receipts: verifier " - f"{html.escape(_bool_code(verifier_written))} / KM " - f"{html.escape(_bool_code(learning_written))}" + "🤖 Agent: " + f"{html.escape(executor_agent)} → " + "independent verifier" ), - "Runtime apply: 1", - f"Next: {html.escape(next_step)}", - f"Runs: {html.escape(incident_runs_url(incident_id, project_id=project_id or 'awoooi'))}", + "🧠 LLM: no runtime apply authority", + ( + "🧾 證據: verifier " + f"{html.escape(_bool_code(verifier_written))} " + f"| KM {html.escape(_bool_code(learning_written))} " + f"| RAG {html.escape(_bool_code(rag_written))} " + f"| MCP {html.escape(_bool_code(mcp_context_ready))} " + "| PlayBook " + f"{html.escape(_bool_code(playbook_trust_written))}" + ), + ( + "🔗 身分: " + f"Run: {html.escape(str(automation_run_id or '')[:8])} " + f"| Incident: {html.escape(str(incident_id))} " + f"| Apply: {html.escape(str(apply_op_id or '')[:8])}" + ), + ( + "📘 PlayBook: " + f"{html.escape(str(playbook_path or '--'))}" + ), + "✍ Runtime apply: 1", + f"⏭ 後續: {html.escape(next_step)}", ] payload: dict = { "text": "\n".join(lines)[:4096], diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index cbdecfdae..99103d7bf 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -888,9 +888,25 @@ def test_terminal_candidate_contract_requires_verified_four_node_chain() -> None async def test_projection_replay_sends_missing_receipt_without_reapplying( monkeypatch: pytest.MonkeyPatch, ) -> None: + closure_coverage = { + "telegram_receipt": False, + "candidate": True, + "check_mode": True, + "controlled_apply": True, + "post_apply_verifier": True, + "auto_repair_execution_receipt": True, + "km_playbook_writeback": True, + "rag_writeback": True, + "mcp_context": True, + "playbook_trust": True, + "timeline_projection": True, + "approval_projection": True, + "execution_lifecycle": True, + "required_runtime_stage_receipts": True, + } readback = AsyncMock( side_effect=[ - {"receipts": {"telegram_receipt": False}}, + {"receipts": closure_coverage}, {"receipts": {"telegram_receipt": True}}, ] ) @@ -937,6 +953,68 @@ async def test_projection_replay_sends_missing_receipt_without_reapplying( assert result["closed"] is True assert result["runtime_apply_executed"] is False telegram_sender.assert_awaited_once() + assert telegram_sender.await_args.kwargs["closure_receipts"] == ( + closure_coverage + ) + + +@pytest.mark.asyncio +async def test_alert_chain_final_receipt_projects_all_learning_evidence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + gateway = AsyncMock() + gateway.send_controlled_apply_result_receipt.return_value = { + "ok": True, + "_awooop_outbound_mirror_acknowledged": True, + } + monkeypatch.setattr( + "src.services.telegram_gateway.get_telegram_gateway", + lambda: gateway, + ) + base = _claim() + claim = replace( + base, + catalog_id="ansible:110-alertmanager-delivery-recovery", + playbook_path=( + "infra/ansible/playbooks/110-alertmanager-delivery-recovery.yml" + ), + apply_playbook_path=( + "infra/ansible/playbooks/110-alertmanager-delivery-recovery.yml" + ), + input_payload={ + **base.input_payload, + "alert_category": "alert_chain_health", + }, + ) + + acknowledged = await service._send_controlled_apply_telegram_receipt( + claim, + _verified_result(), + apply_op_id="00000000-0000-0000-0000-000000000104", + writeback={ + "verification_result": "success", + "verification": True, + "learning": True, + "rag_writeback": True, + "trust_learning": True, + }, + closure_receipts={ + "km_playbook_writeback": True, + "rag_writeback": True, + "mcp_context": True, + "playbook_trust": True, + }, + project_id="awoooi", + ) + + assert acknowledged is True + call = gateway.send_controlled_apply_result_receipt.await_args.kwargs + assert call["alert_category"] == "alert_chain_health" + assert call["verifier_written"] is True + assert call["learning_written"] is True + assert call["rag_written"] is True + assert call["mcp_context_ready"] is True + assert call["playbook_trust_written"] is True @pytest.mark.asyncio @@ -960,10 +1038,11 @@ async def test_projection_stays_open_until_durable_telegram_readback( ] ), ) + telegram_sender = AsyncMock(return_value=False) monkeypatch.setattr( service, "_send_controlled_apply_telegram_receipt", - AsyncMock(return_value=False), + telegram_sender, ) monkeypatch.setattr( service, @@ -994,6 +1073,7 @@ async def test_projection_stays_open_until_durable_telegram_readback( assert result["telegram_receipt_acknowledged"] is False assert result["telegram_lifecycle_written"] is False assert lifecycle.await_count == 1 + telegram_sender.assert_not_awaited() @pytest.mark.asyncio diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py index e3715b42c..2ef2294d0 100644 --- a/apps/api/tests/test_telegram_message_templates.py +++ b/apps/api/tests/test_telegram_message_templates.py @@ -1309,6 +1309,10 @@ async def test_controlled_apply_result_receipt_marks_callback_reply_evidence(mon duration_ms=7727, verifier_written=True, learning_written=True, + rag_written=True, + mcp_context_ready=True, + playbook_trust_written=True, + alert_category="backup_restore", project_id="awoooi", ) @@ -1320,6 +1324,16 @@ async def test_controlled_apply_result_receipt_marks_callback_reply_evidence(mon assert "Run:" in payload["text"] assert "INC-20260627-64472B" in payload["text"] assert "ansible:188-momo-backup-user" in payload["text"] + assert "分類: BACKUP_RESTORE" in payload["text"] + assert "Agent: Host Ansible executor" in payload["text"] + assert "LLM: no runtime apply authority" in payload["text"] + assert ( + "證據: verifier yes | KM yes " + "| RAG yes | MCP yes " + "| PlayBook yes" + ) in payload["text"] + assert payload["text"].count("PlayBook:") == 1 + assert "後續: monitor_for_regression" in payload["text"] source_extra = payload["_awooop_source_envelope_extra"] assert source_extra["callback_reply"]["action"] == "controlled_apply_result" assert source_extra["callback_reply"]["status"] == "callback_reply_sent" @@ -1331,6 +1345,19 @@ async def test_controlled_apply_result_receipt_marks_callback_reply_evidence(mon "00000000-0000-0000-0000-000000000001" ] assert source_extra["outbound_message_type"] == "final" + evidence = source_extra["callback_reply"]["automation_evidence"] + assert evidence == { + "schema_version": "controlled_apply_automation_evidence_v1", + "independent_verifier": True, + "km_writeback": True, + "rag_writeback": True, + "mcp_context": True, + "playbook_trust": True, + "complete": True, + "executor_agent": "Host Ansible executor", + "llm_runtime_apply_authority": False, + } + assert source_extra["callback_reply"]["alert_category"] == "backup_restore" assert ( telegram_gateway_module._infer_outbound_message_type( payload["text"], @@ -1352,6 +1379,40 @@ async def test_controlled_apply_result_receipt_marks_callback_reply_evidence(mon assert snapshot["operator_outcome"]["state"] == "completed_verified" assert snapshot["execution"]["ansible"]["controlled_apply"] is True + sent_requests.clear() + await gateway.send_controlled_apply_result_receipt( + automation_run_id="00000000-0000-0000-0000-000000000011", + incident_id="INC-20260627-64472B", + catalog_id="ansible:110-alertmanager-delivery-recovery", + apply_op_id="73b7a95c-3652-4c0d-bb4c-729e500acedc", + playbook_path=( + "infra/ansible/playbooks/110-alertmanager-delivery-recovery.yml" + ), + verification_result="success", + returncode=0, + duration_ms=880, + verifier_written=True, + learning_written=True, + rag_written=False, + mcp_context_ready=True, + playbook_trust_written=True, + project_id="awoooi", + ) + partial = sent_requests[0][1] + assert "執行完成、證據待補" in partial["text"] + assert "分類: ALERT_CHAIN_HEALTH" in partial["text"] + assert "RAG no" in partial["text"] + assert "complete_missing_evidence_receipts" in partial["text"] + assert ( + partial["_awooop_source_envelope_extra"]["outbound_message_type"] + == "interim" + ) + assert ( + partial["_awooop_source_envelope_extra"]["callback_reply"] + ["automation_evidence"]["complete"] + is False + ) + @pytest.mark.asyncio async def test_no_write_replay_receipt_never_claims_runtime_apply(monkeypatch) -> None: