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

fix(agent99): reconcile durable callbacks as verified repairs

Browse Source
This commit is contained in:
ogt
2026-07-14 17:41:29 +08:00
parent 025829434d
commit 2fe6a56c09
3 changed files with 350 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
apps/api/src/api/v1/platform/events.py
View File

@@ -18,7 +18,11 @@ from src.core.awooop_operator_auth import (
AwoooPOperatorPrincipal,
verify_awooop_operator,
)
from src.core.context import clear_project_context, get_current_project_context, set_project_context
from src.core.context import (
clear_project_context,
get_current_project_context,
set_project_context,
)
from src.services.channel_event_dossier_service import (
RecurrenceWorkItemHandoffKind,
RecurrenceWorkItemMode,
@@ -50,6 +54,7 @@ def _empty_recurrence_summary() -> dict[str, Any]:
"unlinked_event_total": 0,
"auto_repair_linked_total": 0,
"verified_repair_group_total": 0,
"agent99_callback_verified_group_total": 0,
"open_work_item_group_total": 0,
"manual_gate_group_total": 0,
"controlled_apply_gate_group_total": 0,
@@ -239,6 +244,7 @@ class ChannelEventRecurrenceSummary(BaseModel):
unlinked_event_total: int
auto_repair_linked_total: int = 0
verified_repair_group_total: int = 0
agent99_callback_verified_group_total: int = 0
open_work_item_group_total: int = 0
manual_gate_group_total: int = 0
controlled_apply_gate_group_total: int = 0

178
apps/api/src/services/channel_event_dossier_service.py
View File

@@ -33,6 +33,9 @@ _SOURCE_CORRELATION_APPLY_SCHEMA_VERSION = "awooop_source_correlation_apply_v1"
_SOURCE_CORRELATION_APPLY_STAGE = "source_correlation_linked"
_SOURCE_CORRELATION_REVIEW_ACTOR = "awooop_source_correlation_review_service"
_SOURCE_CORRELATION_APPLY_ACTOR = "awooop_source_correlation_apply_service"
_AGENT99_COMPLETION_CALLBACK_ACTOR = "agent99_completion_callback"
_AGENT99_COMPLETION_CALLBACK_SCHEMA = "agent99_completion_callback_v1"
_AGENT99_CALLBACK_REPAIR_EVIDENCE = "agent99_completion_callback"
_INCIDENT_ID_RE = re.compile(r"\bINC-\d{8}-[A-Z0-9]{4,}\b")
RecurrenceWorkItemMode = Literal["auto", "ticket", "reverify", "approval_review", "observe"]
RecurrenceWorkItemHandoffKind = Literal["ticket_proposal", "manual_review"]
@@ -412,7 +415,13 @@ def build_dossier_recurrence(
1
for item in items
if _as_dict(item.get("repair_summary")).get("status")
== "auto_repair_verified"
in {"auto_repair_verified", "agent99_callback_verified"}
),
"agent99_callback_verified_group_total": sum(
1
for item in items
if _as_dict(item.get("repair_summary")).get("status")
== "agent99_callback_verified"
),
"open_work_item_group_total": sum(
1
@@ -464,6 +473,7 @@ def _repair_status(
*,
incident_id: str | None,
latest_run_state: str | None,
latest_stage: str | None,
repair_summary: dict[str, Any] | None,
) -> str:
if not incident_id:
@@ -473,6 +483,18 @@ def _repair_status(
verification = str(
repair_summary.get("latest_verification_result") or ""
).lower()
if (
repair_summary.get("repair_evidence_kind")
== _AGENT99_CALLBACK_REPAIR_EVIDENCE
):
if (
latest_success is True
and verification == "success"
and str(latest_stage or "").lower() == "resolved"
and latest_run_state == "completed"
):
return "agent99_callback_verified"
return "agent99_callback_verification_stale"
if latest_success is True and verification == "success":
return "auto_repair_verified"
if latest_success is True:
@@ -494,6 +516,7 @@ def _work_item_status(repair_status: str) -> str:
return "none"
if repair_status in {
"auto_repair_verified",
"agent99_callback_verified",
"source_correlation_accepted",
"source_correlation_rejected",
}:
@@ -502,6 +525,10 @@ def _work_item_status(repair_status: str) -> str:
def _work_item_kind(repair_status: str, auto_repair_id: Any) -> str:
if repair_status == "agent99_callback_verified":
return "verified_repair"
if repair_status == "agent99_callback_verification_stale":
return "verification"
if repair_status in {
"source_correlation_review",
"source_correlation_accepted",
@@ -526,6 +553,8 @@ def _work_item_next_step(repair_status: str) -> str:
"source_correlation_rejected": "monitor_for_new_provider_evidence",
"source_correlation_applied": "verify_source_link_in_status_chain",
"auto_repair_succeeded_unverified": "run_post_verification",
"agent99_callback_verified": "monitor_recurrence",
"agent99_callback_verification_stale": "rerun_agent99_verifier",
"auto_repair_failed": "triage_failed_repair",
"auto_repair_recorded": "review_repair_record",
"controlled_apply_gate": "evaluate_controlled_apply",
@@ -543,6 +572,8 @@ def _work_item_reason(repair_status: str) -> str:
"source_correlation_rejected": "provider_native_evidence_rejected",
"source_correlation_applied": "provider_native_evidence_link_applied",
"auto_repair_succeeded_unverified": "auto_repair_missing_verification",
"agent99_callback_verified": "agent99_callback_durable_verified",
"agent99_callback_verification_stale": "latest_event_not_verified_resolved",
"auto_repair_failed": "auto_repair_failed",
"auto_repair_recorded": "auto_repair_record_needs_review",
"controlled_apply_gate": "controlled_apply_required",
@@ -573,6 +604,7 @@ def _attach_work_item_summary(
status_value = _repair_status(
incident_id=latest_incident_id,
latest_run_state=group.get("latest_run_state"),
latest_stage=group.get("latest_stage"),
repair_summary=repair_summary,
)
if _needs_source_correlation_review(group, latest_incident_id):
@@ -667,7 +699,10 @@ def _attach_work_item_summary(
"kind": _work_item_kind(status_value, auto_repair_id),
"next_step": work_item_next_step,
"reason": work_item_reason,
"needs_human": work_status == "open",
"needs_human": bool(
work_status == "open"
and status_value != "agent99_callback_verification_stale"
),
}
@@ -2050,6 +2085,133 @@ async def _fetch_auto_repair_summaries_by_incident(
return summaries
async def _fetch_agent99_callback_summaries_by_incident(
db: Any,
incident_ids: list[str],
*,
project_id: str,
) -> dict[str, dict[str, Any]]:
"""Fetch independently verified Agent99 callback receipts by incident."""
visible_incident_ids = incident_ids[:_MAX_REPAIR_INCIDENTS]
if not visible_incident_ids:
return {}
placeholders: list[str] = []
params: dict[str, Any] = {
"actor": _AGENT99_COMPLETION_CALLBACK_ACTOR,
"schema_version": _AGENT99_COMPLETION_CALLBACK_SCHEMA,
"project_id": project_id,
}
for index, incident_id in enumerate(visible_incident_ids):
key = f"agent99_incident_id_{index}"
placeholders.append(f":{key}")
params[key] = incident_id
result = await db.execute(
text(
f"""
WITH callback_events AS (
SELECT
operation.incident_id,
operation.context ->> 'callback_id' AS callback_id,
max(operation.context ->> 'run_id') AS callback_run_id,
bool_or(
operation.event_type = 'EXECUTION_COMPLETED'
AND operation.success IS TRUE
) AS execution_verified,
bool_or(
operation.event_type = 'RESOLVED'
AND operation.success IS TRUE
) AS resolution_verified,
bool_or(
coalesce(operation.context ->> 'verifier_passed', 'false')
= 'true'
) AS verifier_passed,
bool_or(
coalesce(
operation.context ->> 'source_event_resolved',
'false'
) = 'true'
) AS source_event_resolved,
count(*) AS operation_receipt_count,
max(operation.created_at) AS latest_callback_at
FROM alert_operation_log operation
WHERE operation.actor = :actor
AND operation.context ->> 'schema_version' = :schema_version
AND coalesce(
operation.context ->> 'project_id',
:project_id
) = :project_id
AND operation.context ->> 'callback_id' IS NOT NULL
AND operation.incident_id IN ({", ".join(placeholders)})
GROUP BY
operation.incident_id,
operation.context ->> 'callback_id'
),
verified_callbacks AS (
SELECT *
FROM callback_events
WHERE execution_verified IS TRUE
AND resolution_verified IS TRUE
AND verifier_passed IS TRUE
AND source_event_resolved IS TRUE
),
ranked AS (
SELECT
verified_callbacks.*,
count(*) OVER (
PARTITION BY verified_callbacks.incident_id
) AS verified_callback_total,
row_number() OVER (
PARTITION BY verified_callbacks.incident_id
ORDER BY verified_callbacks.latest_callback_at DESC
) AS rn
FROM verified_callbacks
)
SELECT
incident_id,
callback_id,
callback_run_id,
operation_receipt_count,
latest_callback_at,
verified_callback_total
FROM ranked
WHERE rn = 1
"""
),
params,
)
summaries: dict[str, dict[str, Any]] = {}
for row in result.mappings().all():
item = dict(row)
incident_id = str(item.get("incident_id") or "")
if not incident_id:
continue
summaries[incident_id] = {
"schema_version": "awooop_recurrence_repair_summary_v1",
"incident_id": incident_id,
"repair_evidence_kind": _AGENT99_CALLBACK_REPAIR_EVIDENCE,
"latest_auto_repair_id": None,
"latest_agent99_callback_id": item.get("callback_id"),
"latest_agent99_run_id": item.get("callback_run_id"),
"latest_success": True,
"latest_verification_result": "success",
"latest_triggered_by": _AGENT99_COMPLETION_CALLBACK_ACTOR,
"latest_callback_at": item.get("latest_callback_at"),
"operation_receipt_count": int(
item.get("operation_receipt_count") or 0
),
"verified_callback_total": int(
item.get("verified_callback_total") or 0
),
"auto_repair_total": 0,
"success_total": int(item.get("verified_callback_total") or 0),
"failed_total": 0,
}
return summaries
async def _fetch_source_review_decisions_by_work_item(
db: Any,
*,
@@ -2348,10 +2510,20 @@ async def fetch_channel_event_dossier_recurrence(
params,
)
rows = [dict(row) for row in result.mappings().all()]
incident_ids = _collect_incident_ids_from_rows(rows)
repair_summaries = await _fetch_auto_repair_summaries_by_incident(
db,
_collect_incident_ids_from_rows(rows),
incident_ids,
)
agent99_callback_summaries = (
await _fetch_agent99_callback_summaries_by_incident(
db,
incident_ids,
project_id=effective_project_id,
)
)
for incident_id, callback_summary in agent99_callback_summaries.items():
repair_summaries.setdefault(incident_id, callback_summary)
source_review_decisions = await _fetch_source_review_decisions_by_work_item(
db,
project_id=effective_project_id,