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 3a57fed44..8e4e69331 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -637,6 +637,169 @@ async def _record_auto_repair_execution_receipt( return False +def _post_apply_verification_result(result: AnsibleRunResult) -> str: + if result.timed_out: + return "timeout" + return "success" if result.returncode == 0 else "failed" + + +def _post_apply_km_path_type(apply_op_id: str) -> str: + return f"ansible_apply_receipt:{str(apply_op_id or '')[:8]}" + + +def _post_apply_action_label(claim: AnsibleCheckModeClaim, *, apply_op_id: str) -> str: + return ( + f"ansible_controlled_apply:{claim.catalog_id}:" + f"{claim.apply_playbook_path}:apply_op={apply_op_id}" + ) + + +async def _record_post_apply_verifier_and_learning( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + apply_op_id: str, + project_id: str, +) -> dict[str, bool]: + """Persist post-apply verifier evidence and KM learning for Ansible apply.""" + + verification_result = _post_apply_verification_result(result) + action_label = _post_apply_action_label(claim, apply_op_id=apply_op_id) + post_state = { + "schema_version": "ansible_apply_result_verifier_v1", + "verifier": "ansible_apply_result_verifier", + "apply_op_id": apply_op_id, + "catalog_id": claim.catalog_id, + "playbook_path": claim.apply_playbook_path, + "returncode": result.returncode, + "timed_out": result.timed_out, + "action_taken": action_label, + "next_required_step": ( + "km_playbook_trust_writeback" + if verification_result == "success" + else "rollback_or_playbook_repair_candidate" + ), + } + evidence_summary = ( + "AI Agent Ansible controlled apply verifier: " + f"incident={claim.incident_id}; catalog={claim.catalog_id}; " + f"result={verification_result}; returncode={result.returncode}; apply_op={apply_op_id}" + ) + status = {"verification": False, "learning": False} + + try: + async with get_db_context(project_id) as db: + inserted = await db.execute( + text(""" + INSERT INTO incident_evidence ( + id, + incident_id, + matched_playbook_id, + schema_version, + mcp_health, + collection_duration_ms, + sensors_attempted, + sensors_succeeded, + evidence_summary, + post_execution_state, + verification_result + ) + SELECT + gen_random_uuid()::text, + CAST(:incident_id AS varchar(30)), + CAST(:matched_playbook_id AS varchar(36)), + 'v1', + CAST(:mcp_health AS jsonb), + :collection_duration_ms, + 1, + :sensors_succeeded, + :evidence_summary, + CAST(:post_execution_state AS jsonb), + CAST(:verification_result AS varchar(20)) + WHERE NOT EXISTS ( + SELECT 1 + FROM incident_evidence existing + WHERE existing.incident_id = CAST(:incident_id AS varchar(30)) + AND existing.post_execution_state ->> 'apply_op_id' = CAST(:apply_op_id AS text) + ) + RETURNING id + """), + { + "incident_id": claim.incident_id, + "matched_playbook_id": str(claim.catalog_id or "")[:36], + "mcp_health": json.dumps({ + "ansible_apply_result_verifier": verification_result == "success" + }, ensure_ascii=False), + "collection_duration_ms": max(0, int(result.duration_ms or 0)), + "sensors_succeeded": 1 if verification_result == "success" else 0, + "evidence_summary": evidence_summary, + "post_execution_state": json.dumps(post_state, ensure_ascii=False), + "verification_result": verification_result, + "apply_op_id": apply_op_id, + }, + ) + status["verification"] = inserted.scalar() is not None + except Exception as exc: + logger.warning( + "ansible_post_apply_verifier_receipt_failed", + incident_id=claim.incident_id, + catalog_id=claim.catalog_id, + apply_op_id=apply_op_id, + error=str(exc), + ) + + try: + from src.models.knowledge import ( + EntrySource, + EntryStatus, + EntryType, + KnowledgeEntryCreate, + ) + from src.repositories.knowledge_repository import KnowledgeDBRepository + + async with get_db_context(project_id) as db: + repo = KnowledgeDBRepository(db) + await repo.create( + KnowledgeEntryCreate( + title=f"AI 自動修復沉澱：{claim.incident_id}", + content=( + "AI Agent 已完成 Ansible controlled apply 並寫入驗證摘要。\n\n" + f"- Incident: {claim.incident_id}\n" + f"- Catalog: {claim.catalog_id}\n" + f"- PlayBook: {claim.apply_playbook_path}\n" + f"- Apply operation: {apply_op_id}\n" + f"- Verification result: {verification_result}\n" + f"- Return code: {result.returncode}\n" + f"- Next step: {post_state['next_required_step']}\n" + ), + entry_type=EntryType.INCIDENT_CASE, + category="AI自動化/Ansible受控修復", + tags=[ + "ai_auto_repair", + "ansible_controlled_apply", + verification_result, + str(claim.catalog_id or ""), + ], + source=EntrySource.AI_EXTRACTED, + status=EntryStatus.REVIEW, + related_incident_id=claim.incident_id, + related_playbook_id=str(claim.catalog_id or "")[:36] or None, + path_type=_post_apply_km_path_type(apply_op_id), + created_by="ai_agent_ansible_worker", + ) + ) + status["learning"] = True + except Exception as exc: + logger.warning( + "ansible_post_apply_learning_writeback_failed", + incident_id=claim.incident_id, + catalog_id=claim.catalog_id, + apply_op_id=apply_op_id, + error=str(exc), + ) + return status + + async def backfill_missing_auto_repair_execution_receipts_once( *, project_id: str = "awoooi", @@ -645,7 +808,14 @@ async def backfill_missing_auto_repair_execution_receipts_once( ) -> dict[str, Any]: """Create auto_repair_executions receipts for existing Ansible apply rows.""" - stats = {"scanned": 0, "written": 0, "skipped": 0, "error": None} + stats = { + "scanned": 0, + "written": 0, + "verification_written": 0, + "learning_written": 0, + "skipped": 0, + "error": None, + } try: async with get_db_context(project_id) as db: await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) @@ -665,12 +835,32 @@ async def backfill_missing_auto_repair_execution_receipts_once( FROM automation_operation_log apply WHERE apply.operation_type = 'ansible_apply_executed' AND apply.created_at >= NOW() - (:window_hours * INTERVAL '1 hour') - AND NOT EXISTS ( - SELECT 1 - FROM auto_repair_executions existing - WHERE existing.created_at >= NOW() - (:window_hours * INTERVAL '1 hour') - AND existing.triggered_by = 'ansible_controlled_apply' - AND existing.executed_steps::text LIKE '%' || apply.op_id::text || '%' + AND ( + NOT EXISTS ( + SELECT 1 + FROM auto_repair_executions existing + WHERE existing.created_at >= NOW() - (:window_hours * INTERVAL '1 hour') + AND existing.triggered_by = 'ansible_controlled_apply' + AND existing.executed_steps::text LIKE '%' || apply.op_id::text || '%' + ) + OR NOT EXISTS ( + SELECT 1 + FROM incident_evidence evidence + WHERE evidence.incident_id = coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) + AND evidence.post_execution_state ->> 'apply_op_id' = apply.op_id::text + ) + OR NOT EXISTS ( + SELECT 1 + FROM knowledge_entries km + WHERE km.related_incident_id = coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) + AND km.path_type = 'ansible_apply_receipt:' || left(apply.op_id::text, 8) + ) ) ORDER BY apply.created_at DESC LIMIT :limit @@ -694,6 +884,16 @@ async def backfill_missing_auto_repair_execution_receipts_once( stats["written"] += 1 else: stats["skipped"] += 1 + writeback = await _record_post_apply_verifier_and_learning( + claim, + result, + apply_op_id=str(row.get("op_id") or ""), + project_id=project_id, + ) + if writeback.get("verification"): + stats["verification_written"] += 1 + if writeback.get("learning"): + stats["learning_written"] += 1 except Exception as exc: stats["error"] = f"{type(exc).__name__}: {exc}"[:500] logger.warning("ansible_auto_repair_execution_receipt_backfill_failed", **stats) @@ -1111,6 +1311,12 @@ async def run_controlled_apply_for_claim( apply_op_id=apply_op_id, project_id=project_id, ) + writeback = await _record_post_apply_verifier_and_learning( + claim, + result, + apply_op_id=apply_op_id, + project_id=project_id, + ) logger.info( "ansible_controlled_apply_completed", @@ -1122,6 +1328,8 @@ async def run_controlled_apply_for_claim( returncode=result.returncode, timed_out=result.timed_out, auto_repair_receipt_written=receipt_written, + post_apply_verification_written=writeback.get("verification"), + post_apply_learning_written=writeback.get("learning"), ) return result diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 00d101d79..cb085eaf7 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -17,6 +17,8 @@ from src.services.awooop_ansible_check_mode_service import ( _automation_operation_log_incident_id, _build_auto_repair_execution_receipt, _claim_from_apply_operation_row, + _post_apply_km_path_type, + _post_apply_verification_result, _record_auto_repair_execution_receipt, build_ansible_apply_command, build_ansible_check_mode_claim_input, @@ -1529,6 +1531,31 @@ def test_ansible_auto_repair_receipt_insert_casts_asyncpg_parameters() -> None: assert "CAST(:apply_op_id_needle AS text)" in source +def test_ansible_apply_receipt_backfill_includes_verifier_and_km_gaps() -> None: + source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once) + + assert "FROM incident_evidence evidence" in source + assert "evidence.post_execution_state ->> 'apply_op_id' = apply.op_id::text" in source + assert "FROM knowledge_entries km" in source + assert "km.path_type = 'ansible_apply_receipt:' || left(apply.op_id::text, 8)" in source + assert "_record_post_apply_verifier_and_learning" in source + + +def test_ansible_post_apply_verifier_helpers_are_deterministic() -> None: + assert _post_apply_km_path_type("03ca6836-1b76-4da2-8e3e-6d3b6df9254a") == ( + "ansible_apply_receipt:03ca6836" + ) + assert _post_apply_verification_result( + AnsibleRunResult(returncode=0, stdout="", stderr="", duration_ms=1) + ) == "success" + assert _post_apply_verification_result( + AnsibleRunResult(returncode=1, stdout="", stderr="", duration_ms=1) + ) == "failed" + assert _post_apply_verification_result( + AnsibleRunResult(returncode=0, stdout="", stderr="", duration_ms=1, timed_out=True) + ) == "timeout" + + def test_ansible_claim_query_limits_recent_candidate_backlog() -> None: source = inspect.getsource(claim_pending_check_modes)