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

feat(agent99): add durable completion callback

Browse Source
This commit is contained in:
ogt
2026-07-11 20:45:52 +08:00
parent df26f43b77
commit e8b0244ee7
15 changed files with 965 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
apps/api/src/api/v1/agents.py
View File

@@ -26,15 +26,22 @@ Phase 17 技術債修復 (2026-03-26):
"""
import asyncio
import hmac
import json
from typing import Any
from fastapi import APIRouter, BackgroundTasks, HTTPException, status
from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, status
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from src.core.config import settings
from src.core.context import clear_project_context, set_project_context
from src.core.logging import get_logger
from src.core.sse import get_publisher
from src.models.agent99_completion import Agent99CompletionCallbackRequest
from src.services.agent99_completion_callback import (
record_agent99_completion_callback,
)
from src.services.agent99_enterprise_ai_automation_work_items import (
load_latest_agent99_enterprise_ai_automation_work_items,
)
@@ -1173,6 +1180,55 @@ async def get_agent99_enterprise_ai_automation_work_items() -> dict[str, Any]:
) from exc
@router.post(
"/agent99/completion-callback",
response_model=dict[str, Any],
summary="接收 Agent99 已驗證完成回寫",
description=(
"以現有 Agent99 relay token 驗證 99 主機回寫，將同一 callback/run/trace "
"以冪等方式寫入 AWOOOI operation log 與 AwoooP completed shadow run，"
"並在回應前執行 durable DB readback。payload 禁止 raw log、secret 與完整檔案路徑。"
),
)
async def post_agent99_completion_callback(
payload: Agent99CompletionCallbackRequest,
x_agent99_completion_token: str | None = Header(
default=None,
alias="X-Agent99-Completion-Token",
),
) -> dict[str, Any]:
expected_token = settings.AGENT99_SRE_ALERT_RELAY_TOKEN
if not expected_token:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="agent99_completion_auth_not_configured",
)
if not x_agent99_completion_token or not hmac.compare_digest(
x_agent99_completion_token,
expected_token,
):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="agent99_completion_auth_failed",
)
context_tokens = set_project_context(
project_id=payload.project_id,
source="agent99_completion_callback",
request_id=payload.callback_id,
)
try:
receipt = await record_agent99_completion_callback(payload)
finally:
clear_project_context(context_tokens)
if receipt.get("durable_readback") is not True:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="agent99_completion_durable_readback_failed",
)
return receipt
@router.get(
"/awoooi-priority-work-order-readback",
response_model=dict[str, Any],

73
apps/api/src/models/agent99_completion.py Normal file
View File

@@ -0,0 +1,73 @@
"""Bounded Agent99 completion callback contracts."""
from __future__ import annotations
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
class Agent99TelegramReceipt(BaseModel):
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
sent: bool = False
suppressed: bool = False
reason: str | None = Field(default=None, max_length=80)
message_format: str | None = Field(default=None, max_length=80)
visual_sent: bool = False
visual_file: str | None = Field(
default=None,
max_length=180,
pattern=r"^[^/\\]+$",
)
class Agent99ProblemReceipt(BaseModel):
model_config = ConfigDict(extra="forbid")
key: str | None = Field(default=None, max_length=240)
occurrences: int = Field(default=0, ge=0)
verified_resolved: int = Field(default=0, ge=0)
failed: int = Field(default=0, ge=0)
degraded: int = Field(default=0, ge=0)
blocked: int = Field(default=0, ge=0)
verifying: int = Field(default=0, ge=0)
candidate_ready: int = Field(default=0, ge=0)
class Agent99CompletionCallbackRequest(BaseModel):
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
schema_version: Literal["agent99_completion_callback_v1"]
callback_id: str = Field(min_length=3, max_length=180, pattern=r"^[A-Za-z0-9_.:-]+$")
project_id: str = Field(default="awoooi", min_length=1, max_length=64)
run_id: str = Field(min_length=3, max_length=180)
trace_id: str = Field(min_length=3, max_length=180)
work_item_id: str = Field(min_length=3, max_length=180)
alert_id: str | None = Field(default=None, max_length=180)
correlation_key: str | None = Field(default=None, max_length=240)
source: str = Field(default="agent99", max_length=120)
mode: str = Field(min_length=2, max_length=80)
outcome_state: Literal[
"resolved",
"candidate_ready",
"verifying",
"degraded",
"blocked",
"failed",
]
controlled_apply: bool = False
transport_ok: bool
verifier_name: str = Field(min_length=2, max_length=160)
verifier_passed: bool
source_event_resolved: bool
source_event_resolution_policy: str = Field(max_length=120)
duration_seconds: float = Field(ge=0, le=3600)
evidence_ref: str = Field(min_length=1, max_length=180, pattern=r"^[^/\\]+$")
alert_kind: str | None = Field(default=None, max_length=120)
alert_service: str | None = Field(default=None, max_length=160)
alert_host: str | None = Field(default=None, max_length=160)
telegram: Agent99TelegramReceipt = Field(default_factory=Agent99TelegramReceipt)
problem: Agent99ProblemReceipt = Field(default_factory=Agent99ProblemReceipt)
occurred_at: datetime

219
apps/api/src/services/agent99_completion_callback.py Normal file
View File

@@ -0,0 +1,219 @@
"""Durable, idempotent Agent99 completion callback writeback."""
from __future__ import annotations
from typing import Any
from sqlalchemy import text
from src.db.base import get_db_context
from src.models.agent99_completion import Agent99CompletionCallbackRequest
from src.repositories.alert_operation_log_repository import (
get_alert_operation_log_repository,
)
from src.services.channel_hub import (
build_external_alert_provider_event_id,
record_external_alert_event,
)
async def _load_callback_readback(
*,
project_id: str,
provider_event_id: str,
callback_id: str,
) -> dict[str, Any] | None:
async with get_db_context(project_id) as db:
result = await db.execute(
text(
"""
SELECT
event_row.event_id::text AS event_id,
event_row.run_id::text AS run_id,
(
SELECT count(*)
FROM alert_operation_log operation_row
WHERE operation_row.context ->> 'callback_id' = :callback_id
) AS operation_count,
(
SELECT count(*)
FROM alert_operation_log operation_row
WHERE operation_row.context ->> 'callback_id' = :callback_id
AND operation_row.event_type = 'EXECUTION_COMPLETED'
) AS execution_count,
(
SELECT count(*)
FROM alert_operation_log operation_row
WHERE operation_row.context ->> 'callback_id' = :callback_id
AND operation_row.event_type = 'RESOLVED'
) AS resolved_count
FROM awooop_conversation_event event_row
WHERE event_row.project_id = :project_id
AND event_row.channel_type = 'internal'
AND event_row.provider_event_id = :provider_event_id
LIMIT 1
"""
),
{
"project_id": project_id,
"provider_event_id": provider_event_id,
"callback_id": callback_id,
},
)
row = result.mappings().first()
return dict(row) if row else None
def _operation_context(payload: Agent99CompletionCallbackRequest) -> dict[str, Any]:
return {
"schema_version": payload.schema_version,
"callback_id": payload.callback_id,
"project_id": payload.project_id,
"run_id": payload.run_id,
"trace_id": payload.trace_id,
"work_item_id": payload.work_item_id,
"alert_id": payload.alert_id,
"correlation_key": payload.correlation_key,
"source": payload.source,
"mode": payload.mode,
"outcome_state": payload.outcome_state,
"controlled_apply": payload.controlled_apply,
"transport_ok": payload.transport_ok,
"verifier_name": payload.verifier_name,
"verifier_passed": payload.verifier_passed,
"source_event_resolved": payload.source_event_resolved,
"source_event_resolution_policy": payload.source_event_resolution_policy,
"duration_seconds": payload.duration_seconds,
"evidence_ref": payload.evidence_ref,
"alert_kind": payload.alert_kind,
"alert_service": payload.alert_service,
"alert_host": payload.alert_host,
"telegram": payload.telegram.model_dump(mode="json"),
"problem": payload.problem.model_dump(mode="json"),
"occurred_at": payload.occurred_at.isoformat(),
"stores_secret": False,
"raw_log_stored": False,
}
def _owner_plane(alert_kind: str | None) -> str:
value = (alert_kind or "").lower()
if any(marker in value for marker in ("security", "wazuh", "siem", "cve")):
return "iwooos"
return "awoooi"
async def record_agent99_completion_callback(
payload: Agent99CompletionCallbackRequest,
) -> dict[str, Any]:
"""Write one callback to the AWOOOI/AwoooP truth chain and read it back."""
stage = payload.outcome_state
provider_event_id = build_external_alert_provider_event_id(
"agent99",
payload.callback_id,
stage,
)
before = await _load_callback_readback(
project_id=payload.project_id,
provider_event_id=provider_event_id,
callback_id=payload.callback_id,
)
duplicate = before is not None
terminal_resolved = bool(
payload.outcome_state == "resolved"
and payload.verifier_passed
and payload.source_event_resolved
)
severity = (
"info"
if payload.outcome_state == "resolved"
else "critical"
if payload.outcome_state == "failed"
else "warning"
)
context = _operation_context(payload)
event_id = await record_external_alert_event(
project_id=payload.project_id,
provider="agent99",
event_id=payload.callback_id,
stage=stage,
title=f"Agent99 {payload.mode} {payload.outcome_state}",
severity=severity,
namespace="agent99",
target_resource=payload.alert_service or payload.mode,
fingerprint=payload.correlation_key or payload.trace_id,
incident_id=payload.alert_id,
labels={
"agent": "agent99",
"mode": payload.mode,
"outcome_state": payload.outcome_state,
"verifier_passed": str(payload.verifier_passed).lower(),
"source_event_resolved": str(payload.source_event_resolved).lower(),
"owner_plane": _owner_plane(payload.alert_kind),
},
annotations={
"run_id": payload.run_id,
"trace_id": payload.trace_id,
"work_item_id": payload.work_item_id,
"evidence_ref": payload.evidence_ref,
},
payload=context,
is_duplicate=duplicate,
)
repository = get_alert_operation_log_repository()
if int((before or {}).get("execution_count") or 0) == 0:
await repository.append(
"EXECUTION_COMPLETED",
incident_id=payload.alert_id,
actor="agent99_completion_callback",
action_detail=f"{payload.mode}:{payload.outcome_state}",
success=payload.verifier_passed,
context=context,
)
if terminal_resolved and int((before or {}).get("resolved_count") or 0) == 0:
await repository.append(
"RESOLVED",
incident_id=payload.alert_id,
actor="agent99_completion_callback",
action_detail=f"{payload.mode}:verified_resolved",
success=True,
context=context,
)
readback = await _load_callback_readback(
project_id=payload.project_id,
provider_event_id=provider_event_id,
callback_id=payload.callback_id,
)
operation_count = int((readback or {}).get("operation_count") or 0)
execution_count = int((readback or {}).get("execution_count") or 0)
resolved_count = int((readback or {}).get("resolved_count") or 0)
durable_readback = bool(
readback
and str(readback.get("event_id") or "") == str(event_id)
and readback.get("run_id")
and execution_count > 0
and (not terminal_resolved or resolved_count > 0)
)
return {
"schema_version": "agent99_completion_callback_receipt_v1",
"ok": durable_readback,
"status": "duplicate_verified" if duplicate else "recorded_verified",
"accepted": True,
"duplicate": duplicate,
"callback_id": payload.callback_id,
"run_id": payload.run_id,
"trace_id": payload.trace_id,
"work_item_id": payload.work_item_id,
"outcome_state": payload.outcome_state,
"conversation_event_id": str(event_id),
"awooop_run_id": (readback or {}).get("run_id"),
"operation_receipt_count": operation_count,
"execution_receipt_count": execution_count,
"resolved_receipt_count": resolved_count,
"durable_readback": durable_readback,
"stores_secret": False,
"raw_log_stored": False,
}

270
apps/api/tests/test_agent99_completion_callback_api.py Normal file
View File

@@ -0,0 +1,270 @@
from __future__ import annotations
# ruff: noqa: E402
import os
from types import SimpleNamespace
from uuid import UUID
os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import ValidationError
from src.api.v1 import agents as agents_api
from src.core.config import settings
from src.models.agent99_completion import Agent99CompletionCallbackRequest
from src.services import agent99_completion_callback as callback_service
def payload() -> dict:
return {
"schema_version": "agent99_completion_callback_v1",
"callback_id": "agent99-sre-alert-20260711-resolved",
"project_id": "awoooi",
"run_id": "sre-alert-20260711",
"trace_id": "host_recovery:cold-start-gate:no-host",
"work_item_id": "agent99-incident:INC-20260711-001",
"alert_id": "INC-20260711-001",
"correlation_key": "host_recovery:cold-start-gate:no-host",
"source": "agent99-sre-alert-inbox",
"mode": "Recover",
"outcome_state": "resolved",
"controlled_apply": True,
"transport_ok": True,
"verifier_name": "recover_post_condition_v1",
"verifier_passed": True,
"source_event_resolved": True,
"source_event_resolution_policy": "mode_verifier_can_resolve",
"duration_seconds": 9.4,
"evidence_ref": "agent99-Recover-20260711-202342.json",
"alert_kind": "host_recovery",
"alert_service": "cold-start-gate",
"alert_host": "192.168.0.110",
"telegram": {
"sent": True,
"suppressed": False,
"message_format": "incident_card_zh_tw_v4",
"visual_sent": True,
"visual_file": "agent99-card-recovery.png",
},
"problem": {"occurrences": 2, "verified_resolved": 2},
"occurred_at": "2026-07-11T20:23:51+08:00",
}
def app_client() -> TestClient:
app = FastAPI()
app.include_router(agents_api.router, prefix="/api/v1")
return TestClient(app)
def test_completion_callback_fails_closed_without_server_token(monkeypatch) -> None:
monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "")
response = app_client().post(
"/api/v1/agents/agent99/completion-callback",
json=payload(),
)
assert response.status_code == 503
assert response.json()["detail"] == "agent99_completion_auth_not_configured"
def test_completion_payload_rejects_secret_fields_and_full_paths() -> None:
with pytest.raises(ValidationError):
Agent99CompletionCallbackRequest.model_validate(
{**payload(), "bot_token": "must-not-enter-callback"}
)
with pytest.raises(ValidationError):
Agent99CompletionCallbackRequest.model_validate(
{**payload(), "evidence_ref": r"C:\Wooo\Agent99\evidence\raw.json"}
)
def test_completion_callback_rejects_wrong_token(monkeypatch) -> None:
monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected")
response = app_client().post(
"/api/v1/agents/agent99/completion-callback",
headers={"X-Agent99-Completion-Token": "wrong"},
json=payload(),
)
assert response.status_code == 401
assert response.json()["detail"] == "agent99_completion_auth_failed"
def test_completion_callback_requires_durable_readback(monkeypatch) -> None:
monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected")
async def fake_record(_payload):
return {"ok": False, "durable_readback": False}
monkeypatch.setattr(agents_api, "record_agent99_completion_callback", fake_record)
response = app_client().post(
"/api/v1/agents/agent99/completion-callback",
headers={"X-Agent99-Completion-Token": "expected"},
json=payload(),
)
assert response.status_code == 503
assert response.json()["detail"] == "agent99_completion_durable_readback_failed"
def test_completion_callback_returns_same_trace_receipt(monkeypatch) -> None:
monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected")
async def fake_record(request):
return {
"schema_version": "agent99_completion_callback_receipt_v1",
"ok": True,
"durable_readback": True,
"callback_id": request.callback_id,
"run_id": request.run_id,
"trace_id": request.trace_id,
"awooop_run_id": "5e5b0080-e8c9-53a7-872b-0fc1ce80f11f",
}
monkeypatch.setattr(agents_api, "record_agent99_completion_callback", fake_record)
response = app_client().post(
"/api/v1/agents/agent99/completion-callback",
headers={"X-Agent99-Completion-Token": "expected"},
json=payload(),
)
assert response.status_code == 200
assert response.json()["durable_readback"] is True
assert response.json()["callback_id"] == payload()["callback_id"]
assert response.json()["trace_id"] == payload()["trace_id"]
class FakeOperationRepository:
def __init__(self) -> None:
self.events: list[tuple[str, dict]] = []
async def append(self, event_type: str, **kwargs):
self.events.append((event_type, kwargs))
return SimpleNamespace(id=f"operation-{len(self.events)}")
@pytest.mark.asyncio
async def test_completion_service_writes_event_run_and_operation_receipts(monkeypatch) -> None:
request = Agent99CompletionCallbackRequest.model_validate(payload())
event_id = UUID("3c9e8ee7-fb21-5420-afae-ef928f75a58c")
awooop_run_id = "5e5b0080-e8c9-53a7-872b-0fc1ce80f11f"
repository = FakeOperationRepository()
readbacks = iter(
[
None,
{
"event_id": str(event_id),
"run_id": awooop_run_id,
"operation_count": 2,
"execution_count": 1,
"resolved_count": 1,
},
]
)
event_call: dict = {}
async def fake_readback(**_kwargs):
return next(readbacks)
async def fake_record_event(**kwargs):
event_call.update(kwargs)
return event_id
monkeypatch.setattr(callback_service, "_load_callback_readback", fake_readback)
monkeypatch.setattr(callback_service, "record_external_alert_event", fake_record_event)
monkeypatch.setattr(
callback_service,
"get_alert_operation_log_repository",
lambda: repository,
)
receipt = await callback_service.record_agent99_completion_callback(request)
assert receipt["durable_readback"] is True
assert receipt["awooop_run_id"] == awooop_run_id
assert event_call["provider"] == "agent99"
assert event_call["stage"] == "resolved"
assert [item[0] for item in repository.events] == [
"EXECUTION_COMPLETED",
"RESOLVED",
]
assert repository.events[0][1]["context"]["stores_secret"] is False
assert repository.events[0][1]["context"]["raw_log_stored"] is False
@pytest.mark.asyncio
async def test_completion_service_is_idempotent(monkeypatch) -> None:
request = Agent99CompletionCallbackRequest.model_validate(payload())
event_id = UUID("3c9e8ee7-fb21-5420-afae-ef928f75a58c")
existing = {
"event_id": str(event_id),
"run_id": "5e5b0080-e8c9-53a7-872b-0fc1ce80f11f",
"operation_count": 2,
"execution_count": 1,
"resolved_count": 1,
}
repository = FakeOperationRepository()
event_call: dict = {}
async def fake_readback(**_kwargs):
return existing
async def fake_record_event(**kwargs):
event_call.update(kwargs)
return event_id
monkeypatch.setattr(callback_service, "_load_callback_readback", fake_readback)
monkeypatch.setattr(callback_service, "record_external_alert_event", fake_record_event)
monkeypatch.setattr(
callback_service,
"get_alert_operation_log_repository",
lambda: repository,
)
receipt = await callback_service.record_agent99_completion_callback(request)
assert receipt["status"] == "duplicate_verified"
assert receipt["duplicate"] is True
assert event_call["is_duplicate"] is True
assert repository.events == []
@pytest.mark.asyncio
async def test_completion_service_repairs_partial_terminal_receipt(monkeypatch) -> None:
request = Agent99CompletionCallbackRequest.model_validate(payload())
event_id = UUID("3c9e8ee7-fb21-5420-afae-ef928f75a58c")
before = {
"event_id": str(event_id),
"run_id": "5e5b0080-e8c9-53a7-872b-0fc1ce80f11f",
"operation_count": 1,
"execution_count": 1,
"resolved_count": 0,
}
after = {**before, "operation_count": 2, "resolved_count": 1}
readbacks = iter([before, after])
repository = FakeOperationRepository()
async def fake_readback(**_kwargs):
return next(readbacks)
async def fake_record_event(**_kwargs):
return event_id
monkeypatch.setattr(callback_service, "_load_callback_readback", fake_readback)
monkeypatch.setattr(callback_service, "record_external_alert_event", fake_record_event)
monkeypatch.setattr(
callback_service,
"get_alert_operation_log_repository",
lambda: repository,
)
receipt = await callback_service.record_agent99_completion_callback(request)
assert receipt["durable_readback"] is True
assert [item[0] for item in repository.events] == ["RESOLVED"]

14
apps/api/tests/test_agent99_enterprise_ai_automation_work_items_api.py
View File

@@ -31,7 +31,7 @@ def test_agent99_enterprise_work_items_loader_returns_complete_scope() -> None:
"agent99_enterprise_ai_automation_work_items_v1"
)
assert payload["scope_complete"] is False
assert payload["current_p0"]["id"] == "AG99-P0-001"
assert payload["current_p0"]["id"] == "AG99-P0-006"
assert payload["summary"] == {
"host_count": 7,
"product_count": 12,
@@ -40,18 +40,18 @@ def test_agent99_enterprise_work_items_loader_returns_complete_scope() -> None:
"p0_count": 14,
"p1_count": 8,
"p2_count": 3,
"in_progress_count": 10,
"in_progress_count": 9,
"in_progress_blocked_count": 0,
"planned_count": 14,
"complete_count": 1,
"complete_count": 2,
"current_p0_count": 1,
}
assert payload["next_execution_order"][:5] == [
"AG99-P0-001",
"AG99-P0-006",
"AG99-P0-003",
"AG99-P0-004",
"AG99-P0-005",
"AG99-P0-007",
]
assert payload["next_execution_order"][-1] == "AG99-P0-011"
@@ -60,7 +60,7 @@ def test_agent99_enterprise_projection_is_bounded() -> None:
payload = load_latest_agent99_enterprise_ai_automation_work_items()
projection = build_agent99_enterprise_priority_projection(payload)
assert projection["current_p0"]["id"] == "AG99-P0-001"
assert projection["current_p0"]["id"] == "AG99-P0-006"
assert projection["coverage"] == {
"host_count": 7,
"product_count": 12,
@@ -79,7 +79,7 @@ def test_priority_readback_projects_agent99_enterprise_order() -> None:
payload = load_latest_awoooi_priority_work_order_readback()
projection = payload["agent99_enterprise_ai_automation"]
assert projection["current_p0"]["id"] == "AG99-P0-001"
assert projection["current_p0"]["id"] == "AG99-P0-006"
assert projection["work_item_counts"]["p0"] == 14
assert (
payload["source_refs"]["agent99_enterprise_ai_automation_work_items_api"]
@@ -124,7 +124,7 @@ def test_agent99_enterprise_work_items_endpoint_returns_snapshot() -> None:
assert response.status_code == 200
payload = response.json()
assert payload["current_p0"]["id"] == "AG99-P0-001"
assert payload["current_p0"]["id"] == "AG99-P0-006"
assert payload["summary"]["host_count"] == 7
assert payload["summary"]["product_count"] == 12
assert payload["summary"]["public_surface_count"] == 23