From 43ceb3b259ece15aab0b4c648541bc37b02c684e Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 10 Jul 2026 20:00:32 +0800 Subject: [PATCH] feat(agent): bind context and timeline receipts --- .../awooop_ansible_check_mode_service.py | 250 +++++++++++++++++- .../tests/test_awooop_truth_chain_service.py | 83 ++++++ 2 files changed, 332 insertions(+), 1 deletion(-) 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 414a84ee6..c789e517d 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -817,6 +817,59 @@ def build_ansible_pre_apply_runtime_stage_receipts( ] +def build_ansible_context_runtime_stage_receipts( + claim: AnsibleCheckModeClaim, + evidence: dict[str, Any], + *, + derived_from_durable_chain: bool = False, +) -> list[dict[str, Any]]: + """Project public-safe pre-decision evidence into the current run.""" + + evidence_id = str(evidence.get("id") or "") + if not evidence_id: + return [] + + receipts: list[dict[str, Any]] = [] + mcp_health = _json_loads(evidence.get("mcp_health")) + sensors_attempted = max(0, _int_from_value(evidence.get("sensors_attempted"), default=0)) + sensors_succeeded = max(0, _int_from_value(evidence.get("sensors_succeeded"), default=0)) + if mcp_health or sensors_attempted > 0: + receipts.append( + _runtime_stage_receipt( + claim, + stage_id="mcp_context", + evidence_ref=f"incident_evidence:{evidence_id}:mcp_health", + detail={ + "evidence_id": evidence_id, + "tool_status_count": len(mcp_health), + "sensors_attempted": sensors_attempted, + "sensors_succeeded": sensors_succeeded, + "context_collected_before_candidate": True, + }, + derived_from_durable_chain=derived_from_durable_chain, + ) + ) + + recent_logs = evidence.get("recent_logs") + if isinstance(recent_logs, str) and recent_logs.strip(): + receipts.append( + _runtime_stage_receipt( + claim, + stage_id="service_log_evidence", + evidence_ref=f"incident_evidence:{evidence_id}:recent_logs", + detail={ + "evidence_id": evidence_id, + "sanitized_character_count": len(recent_logs), + "content_in_receipt": False, + "redaction_contract": "incident_evidence.recent_logs_sanitized", + "context_collected_before_candidate": True, + }, + derived_from_durable_chain=derived_from_durable_chain, + ) + ) + return receipts + + def build_ansible_post_apply_runtime_stage_receipts( claim: AnsibleCheckModeClaim, result: AnsibleRunResult, @@ -859,6 +912,165 @@ def build_ansible_post_apply_runtime_stage_receipts( return receipts +def build_ansible_timeline_runtime_stage_receipt( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + apply_op_id: str, + timeline_event_id: str, + derived_from_durable_chain: bool = False, +) -> dict[str, Any]: + return _runtime_stage_receipt( + claim, + stage_id="timeline_projection", + evidence_ref=f"timeline_events:{timeline_event_id}", + detail={ + "timeline_event_id": timeline_event_id, + "apply_op_id": apply_op_id, + "returncode": result.returncode, + "projection_status": "success" if result.returncode == 0 else "error", + }, + derived_from_durable_chain=derived_from_durable_chain, + ) + + +async def _load_pre_decision_context_runtime_stage_receipts( + claim: AnsibleCheckModeClaim, + *, + project_id: str, + derived_from_durable_chain: bool = False, +) -> list[dict[str, Any]]: + """Load only evidence collected before the root candidate was created.""" + + try: + async with get_db_context(project_id) as db: + result = await db.execute( + text(""" + SELECT + evidence.id, + evidence.recent_logs, + evidence.mcp_health, + evidence.sensors_attempted, + evidence.sensors_succeeded, + evidence.collected_at + FROM incident_evidence evidence + JOIN automation_operation_log candidate + ON candidate.op_id = CAST(:candidate_op_id AS uuid) + WHERE evidence.incident_id = CAST(:incident_id AS varchar(30)) + AND evidence.collected_at <= candidate.created_at + AND evidence.post_execution_state IS NULL + AND ( + NULLIF(evidence.recent_logs, '') IS NOT NULL + OR evidence.mcp_health IS NOT NULL + ) + ORDER BY evidence.collected_at DESC + LIMIT 10 + """), + { + "candidate_op_id": claim.source_candidate_op_id, + "incident_id": claim.incident_id, + }, + ) + evidence_rows = [dict(row) for row in result.mappings().all()] + except Exception as exc: + logger.warning( + "ansible_pre_decision_context_receipt_read_failed", + incident_id=claim.incident_id, + candidate_op_id=claim.source_candidate_op_id, + error=str(exc), + ) + return [] + + by_stage: dict[str, dict[str, Any]] = {} + for evidence in evidence_rows: + for receipt in build_ansible_context_runtime_stage_receipts( + claim, + evidence, + derived_from_durable_chain=derived_from_durable_chain, + ): + by_stage.setdefault(str(receipt["stage_id"]), receipt) + return list(by_stage.values()) + + +async def _record_timeline_projection_receipt( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + apply_op_id: str, + project_id: str, + derived_from_durable_chain: bool = False, +) -> dict[str, Any] | None: + """Persist one idempotent operator timeline event and return its receipt.""" + + automation_run_id = _automation_run_id_for_claim(claim) + description = ( + f"automation_run_id={automation_run_id};apply_op_id={apply_op_id};" + f"catalog_id={claim.catalog_id};returncode={result.returncode}" + ) + try: + async with get_db_context(project_id) as db: + row = await db.execute( + text(""" + WITH existing AS ( + SELECT id + FROM timeline_events + WHERE incident_id = :incident_id + AND actor = 'ansible_controlled_apply_worker' + AND description = :description + ORDER BY created_at DESC + LIMIT 1 + ), inserted AS ( + INSERT INTO timeline_events ( + incident_id, event_type, status, title, description, + actor, actor_role, risk_level, created_at + ) + SELECT + :incident_id, + 'exec', + :status, + :title, + :description, + 'ansible_controlled_apply_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, + "status": "success" if result.returncode == 0 else "error", + "title": f"AI controlled apply: {claim.catalog_id}"[:500], + "description": description, + "risk_level": str(claim.risk_level or "")[:20] or None, + }, + ) + timeline_event_id = str(row.scalar() or "") + except Exception as exc: + logger.warning( + "ansible_timeline_projection_write_failed", + incident_id=claim.incident_id, + apply_op_id=apply_op_id, + error=str(exc), + ) + return None + + if not timeline_event_id: + return None + return build_ansible_timeline_runtime_stage_receipt( + claim, + result, + apply_op_id=apply_op_id, + timeline_event_id=timeline_event_id, + derived_from_durable_chain=derived_from_durable_chain, + ) + + async def _record_runtime_stage_receipts( claim: AnsibleCheckModeClaim, result: AnsibleRunResult, @@ -867,8 +1079,15 @@ async def _record_runtime_stage_receipts( verifier_ready: bool, project_id: str, derived_from_durable_chain: bool = False, + extra_receipts: tuple[dict[str, Any], ...] = (), ) -> bool: + context_receipts = await _load_pre_decision_context_runtime_stage_receipts( + claim, + project_id=project_id, + derived_from_durable_chain=derived_from_durable_chain, + ) receipts = [ + *context_receipts, *build_ansible_pre_apply_runtime_stage_receipts( claim, derived_from_durable_chain=derived_from_durable_chain, @@ -880,6 +1099,7 @@ async def _record_runtime_stage_receipts( verifier_ready=verifier_ready, derived_from_durable_chain=derived_from_durable_chain, ), + *extra_receipts, ] try: async with get_db_context(project_id) as db: @@ -1328,6 +1548,7 @@ async def backfill_missing_auto_repair_execution_receipts_once( "verification_written": 0, "learning_written": 0, "trust_learning_written": 0, + "timeline_projection_written": 0, "runtime_stage_receipts_written": 0, "skipped": 0, "error": None, @@ -1393,7 +1614,16 @@ async def backfill_missing_auto_repair_execution_receipts_once( WHERE learning.operation_type = 'ansible_learning_writeback_recorded' AND learning.parent_op_id = apply.op_id ) - OR NOT (coalesce(apply.input, '{}'::jsonb) ? 'runtime_stage_receipts') + OR NOT EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + apply.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) + ) AS receipt(value) + WHERE receipt.value ->> 'stage_id' = 'timeline_projection' + ) ) ORDER BY apply.created_at DESC LIMIT :limit @@ -1429,6 +1659,15 @@ async def backfill_missing_auto_repair_execution_receipts_once( stats["learning_written"] += 1 if writeback.get("trust_learning"): stats["trust_learning_written"] += 1 + timeline_receipt = await _record_timeline_projection_receipt( + claim, + result, + apply_op_id=str(row.get("op_id") or ""), + project_id=project_id, + derived_from_durable_chain=True, + ) + if timeline_receipt is not None: + stats["timeline_projection_written"] += 1 if await _record_runtime_stage_receipts( claim, result, @@ -1438,6 +1677,7 @@ 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 (), ): stats["runtime_stage_receipts_written"] += 1 except Exception as exc: @@ -2206,12 +2446,19 @@ async def run_controlled_apply_for_claim( apply_op_id=apply_op_id, project_id=project_id, ) + timeline_receipt = await _record_timeline_projection_receipt( + claim, + result, + apply_op_id=apply_op_id, + project_id=project_id, + ) 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 (), ) telegram_receipt_sent = await _send_controlled_apply_telegram_receipt( claim, @@ -2233,6 +2480,7 @@ 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"), + 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/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 1e3859945..b4a03a4b0 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -19,20 +19,24 @@ from src.services.awooop_ansible_check_mode_service import ( _build_auto_repair_execution_receipt, _claim_from_apply_operation_row, _claim_from_stale_check_mode_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_retry_runtime_stage_receipt, _record_runtime_stage_receipts, + _record_timeline_projection_receipt, _run_ansible_command, _send_controlled_apply_telegram_receipt, backfill_missing_auto_repair_execution_receipts_once, build_ansible_apply_command, build_ansible_check_mode_claim_input, build_ansible_check_mode_command, + build_ansible_context_runtime_stage_receipts, build_ansible_post_apply_runtime_stage_receipts, build_ansible_pre_apply_runtime_stage_receipts, + build_ansible_timeline_runtime_stage_receipt, claim_pending_check_modes, claim_stale_pending_check_modes, detect_ansible_transport_blockers, @@ -1885,6 +1889,85 @@ def test_ansible_runtime_stage_receipts_are_same_run_and_public_safe() -> None: assert inspect.iscoroutinefunction(_record_runtime_stage_receipts) +def test_ansible_context_receipts_reference_pre_decision_evidence_without_content() -> None: + claim = AnsibleCheckModeClaim( + op_id="00000000-0000-0000-0000-000000000112", + source_candidate_op_id="00000000-0000-0000-0000-000000000111", + incident_id="INC-RUNTIME-CONTEXT", + catalog_id="ansible:188-ai-web", + playbook_path="infra/ansible/playbooks/188-ai-web-readonly.yml", + apply_playbook_path="infra/ansible/playbooks/188-ai-web.yml", + inventory_hosts=("host_188",), + risk_level="medium", + input_payload={ + "automation_run_id": "00000000-0000-0000-0000-000000000111", + }, + ) + receipts = build_ansible_context_runtime_stage_receipts( + claim, + { + "id": "evidence-111", + "recent_logs": "sanitized service log summary", + "mcp_health": {"prometheus": True, "kubernetes": False}, + "sensors_attempted": 2, + "sensors_succeeded": 1, + }, + ) + + assert {receipt["stage_id"] for receipt in receipts} == { + "mcp_context", + "service_log_evidence", + } + assert all(receipt["automation_run_id"] == claim.source_candidate_op_id for receipt in receipts) + assert all(receipt["raw_log_payload_stored"] is False for receipt in receipts) + service_log = next( + receipt for receipt in receipts if receipt["stage_id"] == "service_log_evidence" + ) + assert service_log["detail"]["content_in_receipt"] is False + assert "sanitized service log summary" not in str(service_log) + + loader_source = inspect.getsource(_load_pre_decision_context_runtime_stage_receipts) + assert "evidence.collected_at <= candidate.created_at" in loader_source + assert "evidence.post_execution_state IS NULL" in loader_source + + +def test_ansible_timeline_projection_receipt_is_same_run_and_idempotent() -> None: + claim = AnsibleCheckModeClaim( + op_id="00000000-0000-0000-0000-000000000122", + source_candidate_op_id="00000000-0000-0000-0000-000000000121", + incident_id="INC-RUNTIME-TIMELINE", + catalog_id="ansible:188-ai-web", + playbook_path="infra/ansible/playbooks/188-ai-web-readonly.yml", + apply_playbook_path="infra/ansible/playbooks/188-ai-web.yml", + inventory_hosts=("host_188",), + risk_level="medium", + input_payload={ + "automation_run_id": "00000000-0000-0000-0000-000000000121", + }, + ) + result = AnsibleRunResult( + returncode=2, + stdout="", + stderr="apply failed", + duration_ms=42, + ) + receipt = build_ansible_timeline_runtime_stage_receipt( + claim, + result, + apply_op_id="00000000-0000-0000-0000-000000000123", + timeline_event_id="timeline-123", + ) + + assert receipt["stage_id"] == "timeline_projection" + assert receipt["automation_run_id"] == claim.source_candidate_op_id + assert receipt["evidence_ref"] == "timeline_events:timeline-123" + assert receipt["detail"]["projection_status"] == "error" + writer_source = inspect.getsource(_record_timeline_projection_receipt) + assert "WITH existing AS" in writer_source + assert "WHERE NOT EXISTS (SELECT 1 FROM existing)" in writer_source + assert inspect.iscoroutinefunction(_record_timeline_projection_receipt) + + def test_ansible_claim_query_limits_recent_candidate_backlog() -> None: source = inspect.getsource(claim_pending_check_modes)