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

feat(telegram): persist callback status chain snapshots
All checks were successful
CD Pipeline / tests (push) Successful in 1m8s
Details
Code Review / ai-code-review (push) Successful in 11s
Details
CD Pipeline / build-and-deploy (push) Successful in 4m23s
Details
CD Pipeline / post-deploy-checks (push) Successful in 1m31s
Details

Browse Source
This commit is contained in:
Your Name
2026-05-25 09:58:42 +08:00
parent 4818ba45c0
commit daf9d4b00b
8 changed files with 304 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
apps/api/src/services/platform_operator_service.py
View File

@@ -374,6 +374,8 @@ async def list_callback_replies(
m.sent_at,
m.triggered_by_state,
m.source_envelope -> 'callback_reply' AS callback_reply,
m.source_envelope -> 'awooop_status_chain'
AS persisted_awooop_status_chain,
m.source_envelope -> 'km_stale_completion_summary'
AS persisted_km_stale_completion_summary,
r.agent_id,
@@ -1055,6 +1057,9 @@ def _callback_reply_event_item(row: Mapping[str, Any]) -> dict[str, Any]:
"agent_id": row.get("agent_id"),
"run_created_at": row.get("run_created_at"),
"callback_reply": callback_reply,
"persisted_awooop_status_chain": _as_dict(
row.get("persisted_awooop_status_chain"),
) or None,
"persisted_km_stale_completion_summary": _as_dict(
row.get("persisted_km_stale_completion_summary"),
) or None,

150
apps/api/src/services/telegram_gateway.py
View File

@@ -825,6 +825,7 @@ def _callback_reply_source_envelope_extra(
parse_mode: str | None = None,
error: str | None = None,
km_stale_completion_summary: dict[str, object] | None = None,
awooop_status_chain: dict[str, object] | None = None,
) -> dict[str, object] | None:
"""Build AwoooP metadata for Telegram detail/history callback replies."""
if not incident_id:
@@ -856,6 +857,8 @@ def _callback_reply_source_envelope_extra(
)
if km_snapshot:
extra["km_stale_completion_summary"] = km_snapshot
if isinstance(awooop_status_chain, dict):
extra["awooop_status_chain"] = awooop_status_chain
return extra
@@ -909,6 +912,131 @@ def _callback_reply_km_stale_completion_snapshot(
return snapshot
def _callback_reply_awooop_status_chain_snapshot(
*,
incident_id: str | None,
truth_chain: dict[str, object] | None = None,
remediation_history: dict[str, object] | None = None,
) -> dict[str, object] | None:
"""Persist a compact AwoooP status-chain snapshot with callback evidence."""
if not incident_id or (not truth_chain and not remediation_history):
return None
truth_status = (
truth_chain.get("truth_status")
if isinstance(truth_chain, dict) and isinstance(truth_chain.get("truth_status"), dict)
else {}
)
quality = (
truth_chain.get("automation_quality")
if isinstance(truth_chain, dict) and isinstance(truth_chain.get("automation_quality"), dict)
else {}
)
facts = quality.get("facts") if isinstance(quality.get("facts"), dict) else {}
latest = _latest_remediation_history_item(remediation_history)
remediation_state = _remediation_evidence_state(remediation_history) or "missing"
remediation_total = (
_safe_int(remediation_history.get("total"))
if isinstance(remediation_history, dict)
else 0
)
latest_route = "none"
if latest:
latest_route = (
f"{latest.get('agent_id') or 'unknown_agent'}/"
f"{latest.get('tool_name') or 'current_state'}/"
f"{latest.get('required_scope') or 'unknown'}"
)
current_stage = str(truth_status.get("current_stage") or "unknown")
stage_status = str(truth_status.get("stage_status") or "unknown")
verdict = str(quality.get("verdict") or "unknown")
verification = (
facts.get("verification_result")
or latest.get("verification_result_preview")
or "missing"
)
auto_repair_records = _safe_int(facts.get("auto_repair_execution_records"))
operation_records = _safe_int(facts.get("automation_operation_records"))
gateway_total = _safe_int(facts.get("mcp_gateway_total"))
km_entries = _safe_int(facts.get("knowledge_entries"))
needs_human = bool(truth_status.get("needs_human"))
if verdict == "auto_repaired_verified":
repair_state = "auto_repaired_verified"
next_step = "monitor_for_regression"
elif auto_repair_records > 0 or operation_records > 0:
repair_state = "executed_pending_verification" if verification == "missing" else "executed"
next_step = "verify_execution_result"
elif remediation_state == "read_only":
repair_state = "read_only_dry_run"
next_step = "approve_or_escalate_from_awooop"
elif remediation_state == "write_observed":
repair_state = "write_observed_manual_review"
next_step = "review_write_evidence"
elif remediation_state == "blocked":
repair_state = "blocked_manual_required"
next_step = "manual_investigation"
elif needs_human:
repair_state = "manual_required"
next_step = "manual_investigation"
else:
repair_state = "no_execution_evidence"
next_step = "collect_evidence_or_wait"
if remediation_state in {"blocked", "fetch_failed"}:
needs_human = True
if (
remediation_state == "write_observed"
and repair_state != "auto_repaired_verified"
):
needs_human = True
truth_blockers = (
truth_status.get("blockers") if isinstance(truth_status.get("blockers"), list) else []
)
quality_blockers = (
quality.get("blockers") if isinstance(quality.get("blockers"), list) else []
)
blockers = [
str(item)
for item in [*truth_blockers, *quality_blockers]
if item
]
return {
"schema_version": "awooop_status_chain_callback_reply_snapshot_v1",
"source_schema_version": "awooop_status_chain_v1",
"source": "telegram_callback_reply_snapshot",
"source_id": incident_id,
"incident_ids": [incident_id],
"current_stage": current_stage,
"stage_status": stage_status,
"verdict": verdict,
"repair_state": repair_state,
"verification": str(verification),
"needs_human": needs_human,
"next_step": next_step,
"blockers": blockers[:8],
"evidence": {
"auto_repair_records": auto_repair_records,
"operation_records": operation_records,
"mcp_gateway_total": gateway_total,
"knowledge_entries": km_entries,
"remediation_total": remediation_total,
"remediation_state": remediation_state,
"latest_route": latest_route,
"latest_mode": latest.get("mode"),
"latest_at": latest.get("created_at"),
"latest_preview": latest.get("verification_result_preview"),
},
"writes": {
"incident": latest.get("writes_incident_state"),
"auto_repair": latest.get("writes_auto_repair_result"),
},
}
def _merge_outbound_source_envelope_extra(
envelope: dict[str, object],
extra: dict[str, object] | None,
@@ -925,6 +1053,10 @@ def _merge_outbound_source_envelope_extra(
if isinstance(km_stale_completion_summary, dict):
envelope["km_stale_completion_summary"] = km_stale_completion_summary
awooop_status_chain = extra.get("awooop_status_chain")
if isinstance(awooop_status_chain, dict):
envelope["awooop_status_chain"] = awooop_status_chain
extra_refs = extra.get("source_refs")
if isinstance(extra_refs, dict):
source_refs = envelope.setdefault("source_refs", {})
@@ -6123,6 +6255,7 @@ class TelegramGateway:
incident_id=incident_id,
project_id=project_id,
)
truth_chain: dict[str, object] | None = None
try:
from src.services.awooop_truth_chain_service import fetch_truth_chain
@@ -6156,6 +6289,11 @@ class TelegramGateway:
lines += _format_km_stale_completion_lines(km_completion_summary)
lines += _format_remediation_history_lines(remediation_history)
awooop_status_chain_snapshot = _callback_reply_awooop_status_chain_snapshot(
incident_id=incident_id,
truth_chain=truth_chain,
remediation_history=remediation_history,
)
await self._send_html_line_message(
lines,
failure_context="incident_detail",
@@ -6163,6 +6301,7 @@ class TelegramGateway:
incident_id=incident_id,
callback_action="detail",
km_stale_completion_summary=km_completion_summary,
awooop_status_chain=awooop_status_chain_snapshot,
)
except Exception as e:
@@ -6281,6 +6420,7 @@ class TelegramGateway:
incident_id=incident_id,
project_id=project_id,
)
truth_chain: dict[str, object] | None = None
try:
from src.services.awooop_truth_chain_service import fetch_truth_chain
@@ -6309,6 +6449,11 @@ class TelegramGateway:
lines += _format_km_stale_completion_lines(km_completion_summary)
lines += _format_remediation_history_lines(remediation_history)
awooop_status_chain_snapshot = _callback_reply_awooop_status_chain_snapshot(
incident_id=incident_id,
truth_chain=truth_chain,
remediation_history=remediation_history,
)
await self._send_html_line_message(
lines,
failure_context="incident_history",
@@ -6316,6 +6461,7 @@ class TelegramGateway:
incident_id=incident_id,
callback_action="history",
km_stale_completion_summary=km_completion_summary,
awooop_status_chain=awooop_status_chain_snapshot,
)
except Exception as e:
@@ -6555,6 +6701,7 @@ class TelegramGateway:
incident_id: str | None = None,
callback_action: str | None = None,
km_stale_completion_summary: dict[str, object] | None = None,
awooop_status_chain: dict[str, object] | None = None,
) -> None:
"""Send a multi-line HTML message without cutting Telegram tags in half."""
chunks = _telegram_html_chunks(lines)
@@ -6575,6 +6722,7 @@ class TelegramGateway:
callback_action=callback_action,
parse_mode="HTML",
km_stale_completion_summary=km_stale_completion_summary,
awooop_status_chain=awooop_status_chain,
)
if source_extra:
payload[_AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY] = source_extra
@@ -6606,6 +6754,7 @@ class TelegramGateway:
parse_mode="plain_text",
error=str(exc),
km_stale_completion_summary=km_stale_completion_summary,
awooop_status_chain=awooop_status_chain,
)
if fallback_source_extra:
fallback_payload[_AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY] = fallback_source_extra
@@ -6639,6 +6788,7 @@ class TelegramGateway:
parse_mode="plain_text",
error=str(fallback_exc),
km_stale_completion_summary=km_stale_completion_summary,
awooop_status_chain=awooop_status_chain,
)
if rescue_source_extra:
rescue_payload[_AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY] = rescue_source_extra