From 0bb5e9c95aac428113f177bb67f911fee1ce6758 Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 10 Jul 2026 20:19:51 +0800 Subject: [PATCH] feat(agent): persist RAG and PlayBook trust receipts --- .../awooop_ansible_check_mode_service.py | 228 ++++++----- .../awooop_ansible_learning_writeback.py | 368 ++++++++++++++++++ apps/api/src/services/knowledge_service.py | 36 +- .../tests/test_awooop_truth_chain_service.py | 45 ++- 4 files changed, 558 insertions(+), 119 deletions(-) create mode 100644 apps/api/src/services/awooop_ansible_learning_writeback.py 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 052df2e3d..57aea87f3 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -26,6 +26,11 @@ from src.services.ai_automation_runtime_contract import ( AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION, ) from src.services.awooop_ansible_audit_service import get_ansible_catalog_item +from src.services.awooop_ansible_learning_writeback import ( + canonical_ansible_playbook_id, + ensure_ansible_rag_writeback, + record_ansible_playbook_trust_writeback, +) logger = structlog.get_logger(__name__) @@ -1236,107 +1241,20 @@ async def _record_learning_writeback_receipt( verification_result: str, action_label: str, project_id: str, -) -> bool: - """Persist the post-verifier learning receipt after LearningService accepts it.""" +) -> dict[str, Any] | None: + """Persist canonical PlayBook trust and its idempotent learning receipt.""" - matched_playbook_id = str(claim.catalog_id or "")[:36] or None - try: - from src.services.learning_service import get_learning_service - - await get_learning_service().record_verification_result( - incident_id=claim.incident_id, - action_taken=action_label, - verification_result=verification_result, - matched_playbook_id=matched_playbook_id, - ) - except Exception as exc: - logger.warning( - "ansible_post_apply_trust_learning_writeback_failed", - incident_id=claim.incident_id, - catalog_id=claim.catalog_id, - apply_op_id=apply_op_id, - error=str(exc), - ) - return False - - try: - input_payload = { - "schema_version": "ansible_learning_writeback_receipt_v1", - "automation_run_id": str( - claim.input_payload.get("automation_run_id") - or claim.source_candidate_op_id - ), - "incident_id": claim.incident_id, - "catalog_id": claim.catalog_id, - "playbook_path": claim.apply_playbook_path, - "apply_op_id": apply_op_id, - "verification_result": verification_result, - "matched_playbook_id": matched_playbook_id, - "learning_repository": "repair_result", - "playbook_trust_update_attempted": matched_playbook_id is not None, - "stores_raw_logs": False, - "stores_secret_values": False, - } - output_payload = { - "learning_recorded": True, - "success": verification_result == "success", - "returncode": result.returncode, - "timed_out": result.timed_out, - } - async with get_db_context(project_id) as db: - inserted = await db.execute( - text(""" - INSERT INTO automation_operation_log ( - operation_type, actor, status, incident_id, - input, output, dry_run_result, - parent_op_id, tags - ) - SELECT - 'ansible_learning_writeback_recorded', - 'ansible_controlled_apply_worker', - 'success', - :incident_db_id, - CAST(:input AS jsonb), - CAST(:output AS jsonb), - '{}'::jsonb, - CAST(:parent_op_id AS uuid), - :tags - WHERE NOT EXISTS ( - SELECT 1 - FROM automation_operation_log existing - WHERE existing.operation_type = 'ansible_learning_writeback_recorded' - AND existing.parent_op_id = CAST(:parent_op_id AS uuid) - ) - RETURNING op_id - """), - { - "incident_db_id": _automation_operation_log_incident_id(claim.incident_id), - "input": json.dumps(input_payload, ensure_ascii=False), - "output": json.dumps(output_payload, ensure_ascii=False), - "parent_op_id": apply_op_id, - "tags": [ - "ansible", - "controlled_apply", - "learning_writeback", - "playbook_trust", - "ai_agent_auto_execution", - ( - "automation_run_id:" - f"{claim.input_payload.get('automation_run_id') or claim.source_candidate_op_id}" - ), - ], - }, - ) - return inserted.scalar() is not None - except Exception as exc: - logger.warning( - "ansible_learning_writeback_receipt_failed", - incident_id=claim.incident_id, - catalog_id=claim.catalog_id, - apply_op_id=apply_op_id, - error=str(exc), - ) - return False + del result, action_label + return await record_ansible_playbook_trust_writeback( + project_id=project_id, + automation_run_id=_automation_run_id_for_claim(claim), + incident_id=claim.incident_id, + incident_db_id=_automation_operation_log_incident_id(claim.incident_id), + catalog_id=claim.catalog_id, + playbook_path=claim.apply_playbook_path, + apply_op_id=apply_op_id, + verification_result=verification_result, + ) async def _record_post_apply_verifier_and_learning( @@ -1345,7 +1263,7 @@ async def _record_post_apply_verifier_and_learning( *, apply_op_id: str, project_id: str, -) -> dict[str, bool]: +) -> dict[str, Any]: """Persist post-apply verifier evidence and KM learning for Ansible apply.""" verification_result = _post_apply_verification_result(result) @@ -1374,7 +1292,14 @@ async def _record_post_apply_verifier_and_learning( 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, "trust_learning": False} + canonical_playbook_id = canonical_ansible_playbook_id(claim.catalog_id) + status: dict[str, Any] = { + "verification": False, + "learning": False, + "trust_learning": False, + "rag_writeback": False, + "runtime_stage_receipts": [], + } try: async with get_db_context(project_id) as db: @@ -1415,7 +1340,7 @@ async def _record_post_apply_verifier_and_learning( """), { "incident_id": claim.incident_id, - "matched_playbook_id": str(claim.catalog_id or "")[:36], + "matched_playbook_id": canonical_playbook_id, "mcp_health": json.dumps({ "ansible_apply_result_verifier": verification_result == "success" }, ensure_ascii=False), @@ -1495,7 +1420,7 @@ async def _record_post_apply_verifier_and_learning( source=EntrySource.AI_EXTRACTED, status=EntryStatus.REVIEW, related_incident_id=claim.incident_id, - related_playbook_id=str(claim.catalog_id or "")[:36] or None, + related_playbook_id=canonical_playbook_id, path_type=path_type, created_by="ai_agent_ansible_worker", ) @@ -1509,7 +1434,7 @@ async def _record_post_apply_verifier_and_learning( apply_op_id=apply_op_id, error=str(exc), ) - status["trust_learning"] = await _record_learning_writeback_receipt( + trust_writeback = await _record_learning_writeback_receipt( claim, result, apply_op_id=apply_op_id, @@ -1517,6 +1442,41 @@ async def _record_post_apply_verifier_and_learning( action_label=action_label, project_id=project_id, ) + status["trust_learning"] = trust_writeback is not None + if trust_writeback is not None: + status["runtime_stage_receipts"].append( + _runtime_stage_receipt( + claim, + stage_id="playbook_trust", + evidence_ref=( + "playbooks:" + f"{trust_writeback['canonical_playbook_id']}:trust" + ), + detail=trust_writeback, + ) + ) + rag_writeback = ( + await ensure_ansible_rag_writeback( + project_id=project_id, + incident_id=claim.incident_id, + apply_op_id=apply_op_id, + ) + if status["learning"] + else None + ) + status["rag_writeback"] = rag_writeback is not None + if rag_writeback is not None: + status["runtime_stage_receipts"].append( + _runtime_stage_receipt( + claim, + stage_id="rag_writeback", + evidence_ref=( + "knowledge_entries:" + f"{rag_writeback['knowledge_entry_id']}:embedding" + ), + detail=rag_writeback, + ) + ) return status @@ -1525,7 +1485,7 @@ async def _send_controlled_apply_telegram_receipt( result: AnsibleRunResult, *, apply_op_id: str, - writeback: dict[str, bool], + writeback: dict[str, Any], project_id: str, ) -> bool: try: @@ -1573,6 +1533,8 @@ async def backfill_missing_auto_repair_execution_receipts_once( "verification_written": 0, "learning_written": 0, "trust_learning_written": 0, + "rag_writeback_written": 0, + "playbook_trust_written": 0, "timeline_projection_written": 0, "runtime_stage_receipts_written": 0, "skipped": 0, @@ -1649,6 +1611,35 @@ async def backfill_missing_auto_repair_execution_receipts_once( ) AS receipt(value) WHERE receipt.value ->> 'stage_id' = 'timeline_projection' ) + OR NOT EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + apply.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) AS receipt(value) + WHERE receipt.value ->> 'stage_id' = 'playbook_trust' + ) + OR ( + NOT EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + apply.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) AS receipt(value) + WHERE receipt.value ->> 'stage_id' = 'rag_writeback' + ) + AND ( + apply.input -> 'rag_writeback_attempt' IS NULL + OR NULLIF( + apply.input -> 'rag_writeback_attempt' ->> 'attempted_at', + '' + )::timestamptz <= NOW() - INTERVAL '5 minutes' + ) + ) ) ORDER BY apply.created_at DESC LIMIT :limit @@ -1684,6 +1675,9 @@ async def backfill_missing_auto_repair_execution_receipts_once( stats["learning_written"] += 1 if writeback.get("trust_learning"): stats["trust_learning_written"] += 1 + stats["playbook_trust_written"] += 1 + if writeback.get("rag_writeback"): + stats["rag_writeback_written"] += 1 timeline_receipt = await _record_timeline_projection_receipt( claim, result, @@ -1693,6 +1687,11 @@ async def backfill_missing_auto_repair_execution_receipts_once( ) if timeline_receipt is not None: stats["timeline_projection_written"] += 1 + learning_receipts = tuple( + receipt + for receipt in writeback.get("runtime_stage_receipts", []) + if isinstance(receipt, dict) + ) if await _record_runtime_stage_receipts( claim, result, @@ -1702,7 +1701,10 @@ async def backfill_missing_auto_repair_execution_receipts_once( ), project_id=project_id, derived_from_durable_chain=True, - extra_receipts=(timeline_receipt,) if timeline_receipt else (), + extra_receipts=( + *((timeline_receipt,) if timeline_receipt else ()), + *learning_receipts, + ), ): stats["runtime_stage_receipts_written"] += 1 except Exception as exc: @@ -2477,13 +2479,21 @@ async def run_controlled_apply_for_claim( apply_op_id=apply_op_id, project_id=project_id, ) + learning_receipts = tuple( + receipt + for receipt in writeback.get("runtime_stage_receipts", []) + if isinstance(receipt, dict) + ) runtime_stage_receipts_written = await _record_runtime_stage_receipts( claim, result, apply_op_id=apply_op_id, verifier_ready=bool(writeback.get("verification")), project_id=project_id, - extra_receipts=(timeline_receipt,) if timeline_receipt else (), + extra_receipts=( + *((timeline_receipt,) if timeline_receipt else ()), + *learning_receipts, + ), ) telegram_receipt_sent = await _send_controlled_apply_telegram_receipt( claim, @@ -2505,6 +2515,8 @@ async def run_controlled_apply_for_claim( auto_repair_receipt_written=receipt_written, post_apply_verification_written=writeback.get("verification"), post_apply_learning_written=writeback.get("learning"), + rag_writeback_written=writeback.get("rag_writeback"), + playbook_trust_written=writeback.get("trust_learning"), timeline_projection_written=timeline_receipt is not None, runtime_stage_receipts_written=runtime_stage_receipts_written, telegram_receipt_sent=telegram_receipt_sent, diff --git a/apps/api/src/services/awooop_ansible_learning_writeback.py b/apps/api/src/services/awooop_ansible_learning_writeback.py new file mode 100644 index 000000000..34ef80d75 --- /dev/null +++ b/apps/api/src/services/awooop_ansible_learning_writeback.py @@ -0,0 +1,368 @@ +"""Durable RAG and PlayBook trust writeback for the Ansible executor.""" + +from __future__ import annotations + +import hashlib +import json +import re +from typing import Any + +import structlog +from sqlalchemy import select, text + +from src.db.base import get_db_context +from src.db.models import PlaybookRecord +from src.services.awooop_ansible_audit_service import get_ansible_catalog_item +from src.services.knowledge_service import get_knowledge_service +from src.utils.timezone import now_taipei + +logger = structlog.get_logger(__name__) + +_CANONICAL_PLAYBOOK_PREFIX = "PB-ANSIBLE-" + + +def canonical_ansible_playbook_id(catalog_id: str) -> str: + """Return a stable PlayBook primary key for one allowlisted catalog row.""" + + normalized = re.sub( + r"[^A-Z0-9]+", + "-", + str(catalog_id or "").removeprefix("ansible:").upper(), + ).strip("-") + candidate = f"{_CANONICAL_PLAYBOOK_PREFIX}{normalized or 'UNKNOWN'}" + if len(candidate) <= 32: + return candidate + digest = hashlib.sha256(candidate.encode("utf-8")).hexdigest()[:8].upper() + return f"{candidate[:23].rstrip('-')}-{digest}" + + +def _json_dict(value: Any) -> dict[str, Any]: + if isinstance(value, dict): + return value + if isinstance(value, str): + try: + parsed = json.loads(value) + except json.JSONDecodeError: + return {} + return parsed if isinstance(parsed, dict) else {} + return {} + + +async def record_ansible_playbook_trust_writeback( + *, + project_id: str, + automation_run_id: str, + incident_id: str, + incident_db_id: int | None, + catalog_id: str, + playbook_path: str, + apply_op_id: str, + verification_result: str, +) -> dict[str, Any] | None: + """Atomically update canonical PlayBook trust and its learning receipt.""" + + catalog = get_ansible_catalog_item(catalog_id) + if not catalog: + return None + canonical_id = canonical_ansible_playbook_id(catalog_id) + success = verification_result == "success" + now = now_taipei() + + try: + async with get_db_context(project_id) as db: + existing_receipt = await db.execute( + text(""" + SELECT op_id::text AS op_id, input, output + FROM automation_operation_log + WHERE operation_type = 'ansible_learning_writeback_recorded' + AND parent_op_id = CAST(:apply_op_id AS uuid) + ORDER BY created_at DESC + LIMIT 1 + FOR UPDATE + """), + {"apply_op_id": apply_op_id}, + ) + existing_row = existing_receipt.mappings().first() + existing_output = _json_dict( + existing_row.get("output") if existing_row else None + ) + existing_trust = _json_dict( + existing_output.get("playbook_trust_writeback") + ) + if ( + existing_trust.get("durable_write_acknowledged") is True + and existing_trust.get("canonical_playbook_id") == canonical_id + ): + return existing_trust + + selected = await db.execute( + select(PlaybookRecord) + .where(PlaybookRecord.playbook_id == canonical_id) + .with_for_update() + ) + playbook = selected.scalar_one_or_none() + if playbook is None: + playbook = PlaybookRecord( + playbook_id=canonical_id, + project_id=project_id, + name=f"Ansible catalog: {catalog_id}", + description=( + "Canonical trust record for the allowlisted Ansible " + f"catalog action {playbook_path}." + ), + status="approved", + source="yaml_rule", + symptom_pattern={ + "alert_names": [], + "affected_services": list(catalog.get("domains") or []), + "severity_range": ["P0", "P1", "P2", "P3"], + "label_patterns": {}, + "keywords": list(catalog.get("keywords") or [])[:50], + }, + repair_steps=[ + { + "step_number": 1, + "action_type": "script", + "command": playbook_path, + "expected_result": "independent post-verifier passes", + "rollback_command": None, + "requires_approval": bool( + catalog.get("approval_required") + ), + "risk_level": str( + catalog.get("risk_level") or "medium" + ).upper(), + } + ], + estimated_duration_minutes=5, + source_incident_ids=[incident_id], + version=1, + ai_confidence=1.0, + success_count=0, + failure_count=0, + trust_score=0.3, + approved_by="ansible_catalog_policy", + approved_at=now, + tags=["ansible_catalog", catalog_id], + notes="Generated from the controlled executor allowlist.", + requires_approval_level=( + "critical" + if catalog.get("break_glass_required") is True + else "standard" + if catalog.get("approval_required") is True + else "auto" + ), + stateful_targets=[], + requires_pre_backup=False, + review_required=False, + created_at=now, + updated_at=now, + ) + db.add(playbook) + await db.flush() + + 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 + + trust_writeback = { + "schema_version": "ansible_playbook_trust_writeback_v1", + "canonical_playbook_id": canonical_id, + "catalog_id": catalog_id, + "success_count": playbook.success_count, + "failure_count": playbook.failure_count, + "trust_score": round(float(playbook.trust_score), 6), + "verification_result": verification_result, + "durable_write_acknowledged": True, + "source_table": "playbooks", + } + input_payload = { + "schema_version": "ansible_learning_writeback_receipt_v2", + "automation_run_id": automation_run_id, + "incident_id": incident_id, + "catalog_id": catalog_id, + "canonical_playbook_id": canonical_id, + "playbook_path": playbook_path, + "apply_op_id": apply_op_id, + "verification_result": verification_result, + "learning_repository": "playbooks", + "stores_raw_logs": False, + "stores_secret_values": False, + } + output_payload = { + "learning_recorded": True, + "success": success, + "playbook_trust_writeback": trust_writeback, + } + if existing_row: + await db.execute( + text(""" + UPDATE automation_operation_log + SET status = 'success', + input = coalesce(input, '{}'::jsonb) || CAST(:input AS jsonb), + output = coalesce(output, '{}'::jsonb) || CAST(:output AS jsonb), + error = NULL + WHERE op_id = CAST(:op_id AS uuid) + """), + { + "input": json.dumps(input_payload, ensure_ascii=False), + "output": json.dumps(output_payload, ensure_ascii=False), + "op_id": str(existing_row["op_id"]), + }, + ) + else: + await db.execute( + text(""" + INSERT INTO automation_operation_log ( + operation_type, actor, status, incident_id, + input, output, dry_run_result, parent_op_id, tags + ) VALUES ( + 'ansible_learning_writeback_recorded', + 'ansible_controlled_apply_worker', + 'success', + :incident_db_id, + CAST(:input AS jsonb), + CAST(:output AS jsonb), + '{}'::jsonb, + CAST(:parent_op_id AS uuid), + :tags + ) + """), + { + "incident_db_id": incident_db_id, + "input": json.dumps(input_payload, ensure_ascii=False), + "output": json.dumps(output_payload, ensure_ascii=False), + "parent_op_id": apply_op_id, + "tags": [ + "ansible", + "controlled_apply", + "learning_writeback", + "playbook_trust", + "ai_agent_auto_execution", + f"automation_run_id:{automation_run_id}", + ], + }, + ) + return trust_writeback + except Exception as exc: + logger.warning( + "ansible_canonical_playbook_trust_writeback_failed", + incident_id=incident_id, + catalog_id=catalog_id, + apply_op_id=apply_op_id, + error=str(exc), + ) + return None + + +async def ensure_ansible_rag_writeback( + *, + project_id: str, + incident_id: str, + apply_op_id: str, +) -> 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]}" + try: + async with get_db_context(project_id) as db: + selected = await db.execute( + text(""" + SELECT + id, + title, + content, + embedding IS NOT NULL AS embedding_persisted + FROM knowledge_entries + WHERE related_incident_id = :incident_id + AND path_type = :path_type + LIMIT 1 + """), + {"incident_id": incident_id, "path_type": path_type}, + ) + row = selected.mappings().first() + if not row: + return None + entry_id = str(row["id"]) + if row.get("embedding_persisted") is not True: + saved = await get_knowledge_service().ensure_entry_embedding( + entry_id, + str(row.get("title") or ""), + str(row.get("content") or ""), + project_id=project_id, + ) + if not saved: + await _record_rag_attempt( + project_id=project_id, + apply_op_id=apply_op_id, + status="embedding_not_persisted", + ) + return None + async with get_db_context(project_id) as db: + verified = await db.execute( + text(""" + SELECT embedding IS NOT NULL + FROM knowledge_entries + WHERE id = :entry_id + """), + {"entry_id": entry_id}, + ) + if verified.scalar() is not True: + return None + return { + "schema_version": "ansible_rag_writeback_v1", + "knowledge_entry_id": entry_id, + "embedding_persisted": True, + "source_table": "knowledge_entries", + "retrieval_surface": "knowledge_semantic_search", + "durable_write_acknowledged": True, + } + except Exception as exc: + logger.warning( + "ansible_rag_writeback_failed", + incident_id=incident_id, + apply_op_id=apply_op_id, + error=str(exc), + ) + await _record_rag_attempt( + project_id=project_id, + apply_op_id=apply_op_id, + status=type(exc).__name__, + ) + return None + + +async def _record_rag_attempt( + *, + project_id: str, + apply_op_id: str, + status: str, +) -> None: + try: + async with get_db_context(project_id) as db: + await db.execute( + text(""" + UPDATE automation_operation_log + SET input = coalesce(input, '{}'::jsonb) || jsonb_build_object( + 'rag_writeback_attempt', + jsonb_build_object( + 'attempted_at', NOW(), + 'status', CAST(:status AS text) + ) + ) + WHERE op_id = CAST(:apply_op_id AS uuid) + """), + {"apply_op_id": apply_op_id, "status": status[:80]}, + ) + except Exception as exc: + logger.warning( + "ansible_rag_attempt_receipt_failed", + apply_op_id=apply_op_id, + error=str(exc), + ) diff --git a/apps/api/src/services/knowledge_service.py b/apps/api/src/services/knowledge_service.py index 02df79ace..189a8d07f 100644 --- a/apps/api/src/services/knowledge_service.py +++ b/apps/api/src/services/knowledge_service.py @@ -680,18 +680,44 @@ class KnowledgeService: async def _embed_entry(self, entry_id: str, title: str, content: str) -> None: """背景任務：產生並儲存 embedding""" + await self.ensure_entry_embedding( + entry_id, + title, + content, + ) + + async def ensure_entry_embedding( + self, + entry_id: str, + title: str, + content: str, + *, + project_id: str | None = None, + ) -> bool: + """Synchronously persist and acknowledge one KM embedding.""" + try: - text = f"search_document: {title}\n\n{content[:2000]}" - embedding = await self._embed_svc.embed_text(text) + document = f"search_document: {title}\n\n{content[:2000]}" + embedding = await self._embed_svc.embed_text(document) if not embedding: logger.warning("knowledge_embedding_empty", entry_id=entry_id) - return - async with get_db_context() as db: + return False + context = ( + get_db_context(project_id) + if project_id is not None + else get_db_context() + ) + async with context as db: repo = KnowledgeDBRepository(db) - await repo.save_embedding(entry_id, embedding) + saved = await repo.save_embedding(entry_id, embedding) + if not saved: + logger.warning("knowledge_embedding_not_saved", entry_id=entry_id) + return False logger.info("knowledge_embedding_saved", entry_id=entry_id) + return True except Exception as e: logger.warning("knowledge_embedding_failed", entry_id=entry_id, error=str(e)) + return False async def get_entry(self, entry_id: str) -> KnowledgeEntry | None: """取得知識條目 (view_count +1)""" diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index b5caebe77..b7a650424 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -46,6 +46,11 @@ from src.services.awooop_ansible_check_mode_service import ( run_failed_apply_check_mode_replay_once, run_pending_check_modes_once, ) +from src.services.awooop_ansible_learning_writeback import ( + canonical_ansible_playbook_id, + ensure_ansible_rag_writeback, + record_ansible_playbook_trust_writeback, +) from src.services.awooop_truth_chain_service import ( _ansible_playbook_roots, _ansible_runtime_readiness, @@ -1724,16 +1729,42 @@ def test_ansible_apply_receipt_backfill_includes_verifier_and_km_gaps() -> None: assert "_record_post_apply_verifier_and_learning" in source assert "runtime_stage_receipts" in source assert "_record_runtime_stage_receipts" in source + assert "playbook_trust" in source + assert "rag_writeback" in source + assert "INTERVAL '5 minutes'" in source -def test_ansible_learning_writeback_receipt_records_learning_service_call() -> None: +def test_ansible_learning_writeback_uses_canonical_durable_service() -> None: source = inspect.getsource(_record_learning_writeback_receipt) - assert "record_verification_result" in source - assert "ansible_learning_writeback_recorded" in source - assert "learning_repository" in source - assert "stores_raw_logs" in source - assert "stores_secret_values" in source + assert "record_ansible_playbook_trust_writeback" in source + assert "canonical" in inspect.getsource(record_ansible_playbook_trust_writeback) + assert "FOR UPDATE" in inspect.getsource(record_ansible_playbook_trust_writeback) + assert "durable_write_acknowledged" in inspect.getsource( + record_ansible_playbook_trust_writeback + ) + + +def test_ansible_catalog_ids_map_to_stable_playbook_primary_keys() -> None: + expected = { + "ansible:110-devops": "PB-ANSIBLE-110-DEVOPS", + "ansible:188-momo-backup-user": "PB-ANSIBLE-188-MOMO-BACKUP-USER", + "ansible:188-ai-web": "PB-ANSIBLE-188-AI-WEB", + } + + assert { + catalog_id: canonical_ansible_playbook_id(catalog_id) + for catalog_id in expected + } == expected + assert all(len(playbook_id) <= 32 for playbook_id in expected.values()) + + +def test_ansible_rag_writeback_requires_embedding_readback() -> None: + source = inspect.getsource(ensure_ansible_rag_writeback) + + assert "ensure_entry_embedding" in source + assert "embedding IS NOT NULL" in source + assert "durable_write_acknowledged" in source def test_ansible_post_apply_km_writeback_is_idempotent_for_learning_backfill() -> None: @@ -1747,6 +1778,8 @@ def test_ansible_post_apply_km_writeback_is_idempotent_for_learning_backfill() - assert "FROM knowledge_entries" in source assert "path_type = :path_type" in source assert "KnowledgeDBRepository" in source + assert "stage_id=\"rag_writeback\"" in source + assert "stage_id=\"playbook_trust\"" in source def test_ansible_learning_writeback_operation_type_has_schema_migration() -> None: