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

feat(agent): bind context and timeline receipts
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 1m12s
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-10 20:00:32 +08:00
parent bfdc2658bd
commit 43ceb3b259
2 changed files with 332 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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