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

feat(awooop): surface telegram callback coverage
All checks were successful
CD Pipeline / tests (push) Successful in 1m21s
Details
Code Review / ai-code-review (push) Successful in 13s
Details
CD Pipeline / build-and-deploy (push) Successful in 3m59s
Details
CD Pipeline / post-deploy-checks (push) Successful in 1m20s
Details

Browse Source
This commit is contained in:
Your Name
2026-05-25 16:56:28 +08:00
parent b7ee1f47ff
commit 449c4ac807
7 changed files with 524 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -110,6 +110,9 @@ _KM_STALE_COMPLETION_CALLBACK_SCHEMA_VERSION = (
"km_stale_owner_review_completion_callback_summary_v1"
)
_CALLBACK_EVIDENCE_CAPTURE_STATUS_SCHEMA_VERSION = "callback_evidence_capture_status_v1"
_CALLBACK_REPLY_AUDIT_SUMMARY_SCHEMA_VERSION = (
"telegram_callback_reply_audit_summary_v1"
)
# =============================================================================
# Tenants
@@ -398,6 +401,10 @@ async def list_callback_replies(
total = count_result.scalar_one()
rows_result = await db.execute(list_sql, params)
rows = list(rows_result.mappings().all())
summary = await _fetch_callback_reply_audit_summary(
db,
project_id=project_id or "awoooi",
)
items = [_callback_reply_event_item(row) for row in rows]
status_chain_cache: dict[tuple[str, str], dict[str, Any]] = {}
@@ -454,6 +461,165 @@ async def list_callback_replies(
"total": total,
"page": page,
"per_page": per_page,
"summary": summary,
}
async def _fetch_callback_reply_audit_summary(
db: Any,
*,
project_id: str,
) -> dict[str, Any]:
"""Summarize Telegram outbound mirror and callback evidence capture coverage."""
result = await db.execute(
text("""
SELECT
COUNT(*) AS outbound_total,
COUNT(*) FILTER (
WHERE source_envelope <> '{}'::jsonb
) AS outbound_source_envelope_total,
COUNT(*) FILTER (
WHERE source_envelope ? 'source_refs'
) AS outbound_source_refs_total,
COUNT(*) FILTER (
WHERE COALESCE(
source_envelope #> '{source_refs,incident_ids}',
'[]'::jsonb
) <> '[]'::jsonb
) AS outbound_incident_ref_total,
COUNT(*) FILTER (
WHERE send_status = 'failed'
) AS outbound_failed_total,
COUNT(*) FILTER (
WHERE source_envelope ? 'callback_reply'
) AS callback_total,
COUNT(*) FILTER (
WHERE source_envelope #>> '{callback_reply,status}'
= 'callback_reply_sent'
) AS callback_sent_total,
COUNT(*) FILTER (
WHERE source_envelope #>> '{callback_reply,status}'
= 'callback_reply_fallback_sent'
) AS callback_fallback_total,
COUNT(*) FILTER (
WHERE source_envelope #>> '{callback_reply,status}'
= 'callback_reply_rescue_sent'
) AS callback_rescue_total,
COUNT(*) FILTER (
WHERE source_envelope #>> '{callback_reply,status}'
= 'callback_reply_failed'
) AS callback_failed_total,
COUNT(*) FILTER (
WHERE LOWER(source_envelope #>> '{callback_reply,action}')
= 'detail'
) AS callback_detail_total,
COUNT(*) FILTER (
WHERE LOWER(source_envelope #>> '{callback_reply,action}')
= 'history'
) AS callback_history_total,
COUNT(*) FILTER (
WHERE source_envelope ? 'callback_reply'
AND source_envelope ? 'awooop_status_chain'
AND source_envelope ? 'km_stale_completion_summary'
) AS callback_snapshot_captured_total,
COUNT(*) FILTER (
WHERE source_envelope ? 'callback_reply'
AND (
source_envelope ? 'awooop_status_chain'
OR source_envelope ? 'km_stale_completion_summary'
)
AND NOT (
source_envelope ? 'awooop_status_chain'
AND source_envelope ? 'km_stale_completion_summary'
)
) AS callback_snapshot_partial_total,
COUNT(*) FILTER (
WHERE source_envelope ? 'callback_reply'
AND NOT (
source_envelope ? 'awooop_status_chain'
OR source_envelope ? 'km_stale_completion_summary'
)
) AS callback_snapshot_missing_total,
COUNT(DISTINCT source_envelope #>> '{callback_reply,incident_id}')
FILTER (
WHERE source_envelope ? 'callback_reply'
AND COALESCE(
source_envelope #>> '{callback_reply,incident_id}',
''
) <> ''
) AS callback_incident_total,
MAX(COALESCE(sent_at, queued_at)) AS latest_outbound_at,
MAX(COALESCE(sent_at, queued_at)) FILTER (
WHERE source_envelope ? 'callback_reply'
) AS latest_callback_at
FROM awooop_outbound_message
WHERE project_id = :project_id
AND channel_type = 'telegram'
"""),
{"project_id": project_id},
)
return _callback_reply_audit_summary_from_row(
result.mappings().one(),
project_id=project_id,
)
def _callback_reply_audit_summary_from_row(
row: Mapping[str, Any],
*,
project_id: str,
) -> dict[str, Any]:
"""Convert aggregate SQL row into the public callback evidence audit summary."""
outbound_total = _safe_int(row.get("outbound_total"))
callback_total = _safe_int(row.get("callback_total"))
captured = _safe_int(row.get("callback_snapshot_captured_total"))
partial = _safe_int(row.get("callback_snapshot_partial_total"))
missing = _safe_int(row.get("callback_snapshot_missing_total"))
outbound_incident_refs = _safe_int(row.get("outbound_incident_ref_total"))
if callback_total <= 0:
snapshot_status = "no_callback"
next_action = "press_telegram_detail_or_history"
elif missing > 0:
snapshot_status = "not_captured"
next_action = "press_telegram_detail_or_history_after_rollout"
elif partial > 0:
snapshot_status = "partial"
next_action = "press_telegram_detail_or_history_after_rollout"
elif outbound_total > 0 and outbound_incident_refs == 0:
snapshot_status = "captured"
next_action = "review_outbound_source_refs"
else:
snapshot_status = "captured"
next_action = "none"
return {
"schema_version": _CALLBACK_REPLY_AUDIT_SUMMARY_SCHEMA_VERSION,
"project_id": project_id,
"outbound_total": outbound_total,
"outbound_source_envelope_total": _safe_int(
row.get("outbound_source_envelope_total")
),
"outbound_source_refs_total": _safe_int(
row.get("outbound_source_refs_total")
),
"outbound_incident_ref_total": outbound_incident_refs,
"outbound_failed_total": _safe_int(row.get("outbound_failed_total")),
"callback_total": callback_total,
"callback_sent_total": _safe_int(row.get("callback_sent_total")),
"callback_fallback_total": _safe_int(row.get("callback_fallback_total")),
"callback_rescue_total": _safe_int(row.get("callback_rescue_total")),
"callback_failed_total": _safe_int(row.get("callback_failed_total")),
"callback_detail_total": _safe_int(row.get("callback_detail_total")),
"callback_history_total": _safe_int(row.get("callback_history_total")),
"callback_snapshot_captured_total": captured,
"callback_snapshot_partial_total": partial,
"callback_snapshot_missing_total": missing,
"callback_incident_total": _safe_int(row.get("callback_incident_total")),
"snapshot_status": snapshot_status,
"next_action": next_action,
"latest_outbound_at": row.get("latest_outbound_at"),
"latest_callback_at": row.get("latest_callback_at"),
}