wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(agent): persist RAG and PlayBook trust receipts
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details
CD Pipeline / tests (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-10 20:19:51 +08:00
parent 7079b1e0ec
commit 0bb5e9c95a
4 changed files with 558 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

228
apps/api/src/services/awooop_ansible_check_mode_service.py
View File

@@ -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,

368
apps/api/src/services/awooop_ansible_learning_writeback.py Normal file
View File

@@ -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),
)

36
apps/api/src/services/knowledge_service.py
View File

@@ -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)"""

45
apps/api/tests/test_awooop_truth_chain_service.py
View File

@@ -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: