f4cf465676
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 3m23s
CD Pipeline / build-and-deploy (push) Successful in 14m17s
CD Pipeline / post-deploy-checks (push) Successful in 1m57s
2338 lines
94 KiB
Python
2338 lines
94 KiB
Python
"""Durable single-owner ledger for Agent99 controlled dispatch.
|
|
|
|
One logical recovery action must have one identity across Alertmanager,
|
|
Agent99, Telegram projection, verifier, and learning writeback. This module
|
|
uses the existing AwoooP run/idempotency/step tables as a PostgreSQL-backed
|
|
reservation and receipt store. It never performs the transport dispatch.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
from uuid import NAMESPACE_URL, UUID, uuid4, uuid5
|
|
|
|
import structlog
|
|
from sqlalchemy import and_, or_, select, update
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
|
|
from src.db.awooop_models import (
|
|
AwoooPRunIdempotency,
|
|
AwoooPRunState,
|
|
AwoooPRunStepJournal,
|
|
)
|
|
from src.db.base import get_db_context
|
|
from src.db.models import AlertOperationLog, IncidentRecord
|
|
from src.models.incident import IncidentStatus
|
|
from src.services.agent99_public_receipts import (
|
|
AGENT99_BACKUP_REQUIRED_VERIFIER_EVIDENCE_REFS,
|
|
AGENT99_LEARNING_RECEIPT_REFS,
|
|
AGENT99_REQUIRED_VERIFIER_EVIDENCE_REFS,
|
|
normalize_agent99_public_receipt_ref,
|
|
sanitize_agent99_public_receipt_refs,
|
|
)
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
AGENT99_DISPATCH_AGENT_ID = "agent99_controlled_dispatch"
|
|
AGENT99_DISPATCH_CHANNEL = "agent99_controlled"
|
|
AGENT99_DISPATCH_TRIGGER_TYPE = "agent99_dispatch"
|
|
AGENT99_DISPATCH_LEASE_SECONDS = 120
|
|
AGENT99_DISPATCH_RETRY_SECONDS = 600
|
|
AGENT99_DISPATCH_TIMEOUT_MINUTES = 30
|
|
AGENT99_MAX_EXECUTION_GENERATION = 3
|
|
|
|
|
|
def _utc_now_naive() -> datetime:
|
|
"""Return PostgreSQL-compatible UTC for the existing naive columns."""
|
|
|
|
return datetime.now(UTC).replace(tzinfo=None)
|
|
|
|
|
|
def _stable_json(value: dict[str, Any]) -> str:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def _sha256(value: str) -> str:
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _agent99_exact_nonnegative_int(value: Any) -> int | None:
|
|
"""Normalize a JSON number only when it is an exact non-negative int."""
|
|
|
|
if isinstance(value, bool) or not isinstance(value, int | float):
|
|
return None
|
|
if not math.isfinite(float(value)):
|
|
return None
|
|
normalized = int(value)
|
|
if normalized < 0 or float(value) != float(normalized):
|
|
return None
|
|
return normalized
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Agent99DispatchIdentity:
|
|
project_id: str
|
|
incident_id: str
|
|
source_fingerprint: str
|
|
route_id: str
|
|
execution_generation: str
|
|
approval_id: str
|
|
run_id: UUID
|
|
trace_id: str
|
|
work_item_id: str
|
|
idempotency_key: str
|
|
single_flight_key: str
|
|
lock_owner: str
|
|
trigger_ref: str
|
|
|
|
def public_dict(self) -> dict[str, str]:
|
|
public = {
|
|
"schema_version": "agent99_controlled_dispatch_identity_v1",
|
|
"project_id": self.project_id,
|
|
"incident_id": self.incident_id,
|
|
"source_fingerprint": self.source_fingerprint,
|
|
"route_id": self.route_id,
|
|
"execution_generation": self.execution_generation,
|
|
"approval_id": self.approval_id,
|
|
"run_id": str(self.run_id),
|
|
"trace_id": self.trace_id,
|
|
"work_item_id": self.work_item_id,
|
|
"idempotency_key": self.idempotency_key,
|
|
"single_flight_key": self.single_flight_key,
|
|
"lock_owner": self.lock_owner,
|
|
}
|
|
public["canonical_digest"] = _sha256(_stable_json(public))
|
|
return public
|
|
|
|
|
|
def build_agent99_dispatch_identity(
|
|
*,
|
|
project_id: str,
|
|
incident_id: str,
|
|
source_fingerprint: str,
|
|
route_id: str,
|
|
execution_generation: str = "1",
|
|
approval_id: str = "",
|
|
work_item_id: str = "",
|
|
) -> Agent99DispatchIdentity:
|
|
"""Build one deterministic identity for every projection of a dispatch."""
|
|
|
|
resolved_project_id = str(project_id or "awoooi").strip() or "awoooi"
|
|
resolved_incident_id = str(incident_id or "").strip()
|
|
resolved_route_id = str(route_id or "").strip()
|
|
resolved_work_item_id = (
|
|
str(work_item_id or "").strip()
|
|
or (
|
|
f"agent99-dispatch:{resolved_project_id}:"
|
|
f"{resolved_incident_id}:{resolved_route_id}"
|
|
)
|
|
)[:160]
|
|
normalized = {
|
|
"project_id": resolved_project_id,
|
|
"incident_id": resolved_incident_id,
|
|
"source_fingerprint": str(source_fingerprint or "").strip(),
|
|
"route_id": resolved_route_id,
|
|
"execution_generation": str(execution_generation or "1").strip() or "1",
|
|
"approval_id": str(approval_id or "").strip()[:128],
|
|
"work_item_id": resolved_work_item_id,
|
|
}
|
|
if not normalized["incident_id"]:
|
|
raise ValueError("agent99_dispatch_incident_id_required")
|
|
if not normalized["source_fingerprint"]:
|
|
raise ValueError("agent99_dispatch_source_fingerprint_required")
|
|
if not normalized["route_id"]:
|
|
raise ValueError("agent99_dispatch_route_id_required")
|
|
|
|
canonical = _stable_json(normalized)
|
|
digest = _sha256(canonical)
|
|
# The transport lock protects one logical incident/source/route across
|
|
# approval refreshes and bounded generations. Run/idempotency identity is
|
|
# stricter and includes both approval and work-item identity above.
|
|
single_flight_digest = _sha256(_stable_json({
|
|
"project_id": normalized["project_id"],
|
|
"incident_id": normalized["incident_id"],
|
|
"source_fingerprint": normalized["source_fingerprint"],
|
|
"route_id": normalized["route_id"],
|
|
}))
|
|
run_id = uuid5(NAMESPACE_URL, f"agent99-controlled-dispatch:{canonical}")
|
|
trace_id = f"00-{digest[:32]}-{digest[32:48]}-01"
|
|
idempotency_key = f"agent99-controlled:{digest}"
|
|
trigger_ref = (
|
|
f"agent99:{normalized['incident_id']}:{normalized['route_id']}:"
|
|
f"{normalized['execution_generation']}"
|
|
)[:256]
|
|
return Agent99DispatchIdentity(
|
|
project_id=normalized["project_id"],
|
|
incident_id=normalized["incident_id"],
|
|
source_fingerprint=normalized["source_fingerprint"],
|
|
route_id=normalized["route_id"],
|
|
execution_generation=normalized["execution_generation"],
|
|
approval_id=normalized["approval_id"],
|
|
run_id=run_id,
|
|
trace_id=trace_id,
|
|
work_item_id=normalized["work_item_id"],
|
|
idempotency_key=idempotency_key,
|
|
single_flight_key=f"agent99-controlled-flight:{single_flight_digest}",
|
|
lock_owner=str(run_id),
|
|
trigger_ref=trigger_ref,
|
|
)
|
|
|
|
|
|
def parse_agent99_dispatch_identity(
|
|
value: dict[str, Any],
|
|
) -> Agent99DispatchIdentity:
|
|
"""Rebuild and validate a complete public identity from an outcome caller."""
|
|
|
|
required_nonempty = {
|
|
"schema_version",
|
|
"project_id",
|
|
"incident_id",
|
|
"source_fingerprint",
|
|
"route_id",
|
|
"execution_generation",
|
|
"run_id",
|
|
"trace_id",
|
|
"work_item_id",
|
|
"idempotency_key",
|
|
"single_flight_key",
|
|
"lock_owner",
|
|
"canonical_digest",
|
|
}
|
|
normalized = {
|
|
str(key): str(item or "").strip()
|
|
for key, item in (value or {}).items()
|
|
}
|
|
missing = sorted(
|
|
key for key in required_nonempty if not normalized.get(key)
|
|
)
|
|
if "approval_id" not in normalized:
|
|
missing.append("approval_id")
|
|
if missing:
|
|
raise ValueError(
|
|
"agent99_dispatch_identity_fields_required:" + ",".join(missing)
|
|
)
|
|
if normalized["schema_version"] != "agent99_controlled_dispatch_identity_v1":
|
|
raise ValueError("agent99_dispatch_identity_schema_invalid")
|
|
identity = build_agent99_dispatch_identity(
|
|
project_id=normalized["project_id"],
|
|
incident_id=normalized["incident_id"],
|
|
source_fingerprint=normalized["source_fingerprint"],
|
|
route_id=normalized["route_id"],
|
|
execution_generation=normalized["execution_generation"],
|
|
approval_id=normalized.get("approval_id", ""),
|
|
work_item_id=normalized["work_item_id"],
|
|
)
|
|
expected = identity.public_dict()
|
|
compared = required_nonempty | {"approval_id"}
|
|
mismatched = sorted(
|
|
key for key in (compared - {"canonical_digest"})
|
|
if normalized.get(key, "") != str(expected.get(key) or "")
|
|
)
|
|
if mismatched:
|
|
raise ValueError(
|
|
"agent99_dispatch_identity_mismatch:" + ",".join(mismatched)
|
|
)
|
|
if normalized["canonical_digest"] != expected["canonical_digest"]:
|
|
raise ValueError("agent99_dispatch_identity_mismatch:canonical_digest")
|
|
return identity
|
|
|
|
|
|
def validate_agent99_outcome_identity(
|
|
identity: Agent99DispatchIdentity,
|
|
receipt: dict[str, Any],
|
|
) -> bool:
|
|
"""Require the complete canonical identity at both receipt layers."""
|
|
|
|
outer = receipt.get("identity")
|
|
outcome = receipt.get("outcome")
|
|
nested = outcome.get("identity") if isinstance(outcome, dict) else None
|
|
if not isinstance(outer, dict) or not isinstance(nested, dict):
|
|
return False
|
|
expected = identity.public_dict()
|
|
for candidate in (outer, nested):
|
|
try:
|
|
parsed = parse_agent99_dispatch_identity(candidate)
|
|
except ValueError:
|
|
return False
|
|
if parsed != identity or parsed.public_dict() != expected:
|
|
return False
|
|
return True
|
|
|
|
|
|
def agent99_dispatch_identity_from_public(
|
|
value: dict[str, Any],
|
|
) -> Agent99DispatchIdentity:
|
|
"""Rebuild and cryptographically validate a public dispatch identity."""
|
|
|
|
return parse_agent99_dispatch_identity(value)
|
|
|
|
|
|
def attach_agent99_dispatch_identity(
|
|
payload: dict[str, Any],
|
|
identity: Agent99DispatchIdentity,
|
|
) -> dict[str, Any]:
|
|
"""Attach the shared identity without exposing secrets or raw responses."""
|
|
|
|
enriched = dict(payload)
|
|
routing = dict(enriched.get("routing") or {})
|
|
awoooi = dict(enriched.get("awoooi") or {})
|
|
public_identity = identity.public_dict()
|
|
routing.update({
|
|
"idempotencyKey": identity.idempotency_key,
|
|
"runId": str(identity.run_id),
|
|
"traceId": identity.trace_id,
|
|
"workItemId": identity.work_item_id,
|
|
"executionGeneration": identity.execution_generation,
|
|
"correlationKey": identity.idempotency_key,
|
|
})
|
|
awoooi.update({
|
|
"incidentId": identity.incident_id,
|
|
"approvalId": identity.approval_id,
|
|
"agent99DispatchIdentity": public_identity,
|
|
})
|
|
enriched.update({
|
|
"id": f"awoooi-agent99-{identity.run_id}",
|
|
"routing": routing,
|
|
"awoooi": awoooi,
|
|
})
|
|
return enriched
|
|
|
|
|
|
def build_agent99_dispatch_receipt_envelope(
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
dispatch_receipt: dict[str, Any],
|
|
controlled_apply_authorized: bool | None = None,
|
|
) -> dict[str, Any]:
|
|
accepted = bool(dispatch_receipt.get("accepted") is True)
|
|
inbox_triggered = bool(dispatch_receipt.get("inbox_triggered") is True)
|
|
dispatch_promoted = bool(accepted and inbox_triggered)
|
|
delivery_certainty = str(
|
|
dispatch_receipt.get("delivery_certainty")
|
|
or (
|
|
"delivered"
|
|
if dispatch_promoted
|
|
else "unknown"
|
|
)
|
|
)
|
|
if delivery_certainty not in {"delivered", "not_delivered", "unknown"}:
|
|
delivery_certainty = "unknown"
|
|
retry_safe = bool(
|
|
not dispatch_promoted
|
|
and not accepted
|
|
and delivery_certainty == "not_delivered"
|
|
)
|
|
public_identity = identity.public_dict()
|
|
stage_identity = {
|
|
"run_id": str(identity.run_id),
|
|
"trace_id": identity.trace_id,
|
|
"work_item_id": identity.work_item_id,
|
|
}
|
|
return {
|
|
"schema_version": "agent99_controlled_dispatch_receipt_v1",
|
|
"status": (
|
|
"dispatch_accepted_verifier_pending"
|
|
if dispatch_promoted
|
|
else "dispatch_not_delivered_retry_safe"
|
|
if retry_safe
|
|
else "dispatch_delivery_unknown_reconcile_only"
|
|
),
|
|
"identity": public_identity,
|
|
"dispatch_receipt": {
|
|
"schema_version": str(
|
|
dispatch_receipt.get("schema_version")
|
|
or "agent99_sre_dispatch_receipt_v1"
|
|
),
|
|
"status": str(dispatch_receipt.get("status") or "unknown"),
|
|
"transport": str(dispatch_receipt.get("transport") or "unknown"),
|
|
"alert_id": str(dispatch_receipt.get("alert_id") or ""),
|
|
"kind": str(dispatch_receipt.get("kind") or ""),
|
|
"http_status": int(dispatch_receipt.get("http_status") or 0),
|
|
"accepted": accepted,
|
|
"inbox_triggered": inbox_triggered,
|
|
"delivery_certainty": delivery_certainty,
|
|
"stores_raw_response": False,
|
|
},
|
|
# Transport acceptance proves only that Agent99 accepted the inbox
|
|
# item. Runtime execution stays unproven until the independent verifier
|
|
# and durable learning acknowledgements write into this same run.
|
|
"dispatch_accepted": accepted,
|
|
"inbox_triggered": inbox_triggered,
|
|
"dispatch_promoted": dispatch_promoted,
|
|
"controlled_apply_authorized": bool(
|
|
dispatch_promoted
|
|
and (
|
|
dispatch_receipt.get("controlled_apply_authorized") is True
|
|
if controlled_apply_authorized is None
|
|
else controlled_apply_authorized
|
|
)
|
|
),
|
|
"runtime_execution_authorized": False,
|
|
"runtime_execution_attempted": False,
|
|
"post_verifier_passed": False,
|
|
"runtime_closure_verified": False,
|
|
"closure_complete": False,
|
|
"needs_human": False,
|
|
"retry_policy": {
|
|
"safe_to_retry": retry_safe,
|
|
"requires_generation_advance": retry_safe,
|
|
"reconcile_only": bool(not dispatch_promoted and not retry_safe),
|
|
"max_generation": AGENT99_MAX_EXECUTION_GENERATION,
|
|
},
|
|
"outbox": {
|
|
"status": (
|
|
"delivered"
|
|
if dispatch_promoted
|
|
else "not_delivered"
|
|
if retry_safe
|
|
else "delivery_unknown"
|
|
),
|
|
"durable_reservation": True,
|
|
"step_seq": 1,
|
|
**stage_identity,
|
|
},
|
|
"verifier": {
|
|
"status": (
|
|
"pending"
|
|
if dispatch_promoted
|
|
else "blocked_dispatch_not_promoted"
|
|
),
|
|
"step_seq": 2,
|
|
**stage_identity,
|
|
"required_receipts": [
|
|
"agent99_outcome_terminal_readback",
|
|
"full_stack_cold_start_scorecard_rerun",
|
|
"source_fingerprint_resolution",
|
|
],
|
|
},
|
|
"learning_writeback": {
|
|
"status": (
|
|
"pending_verifier"
|
|
if dispatch_promoted
|
|
else "blocked_dispatch_not_promoted"
|
|
),
|
|
"step_seq": 3,
|
|
**stage_identity,
|
|
"required_receipts": [
|
|
"incident_closure_receipt",
|
|
"telegram_lifecycle_receipt",
|
|
"km_writeback_ack",
|
|
"playbook_trust_writeback_ack",
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def _stage_identity(identity: Agent99DispatchIdentity) -> dict[str, str]:
|
|
return {
|
|
"run_id": str(identity.run_id),
|
|
"trace_id": identity.trace_id,
|
|
"work_item_id": identity.work_item_id,
|
|
}
|
|
|
|
|
|
def _stage_identity_matches(
|
|
identity: Agent99DispatchIdentity,
|
|
receipt: dict[str, Any],
|
|
) -> bool:
|
|
"""Validate one complete canonical identity projection."""
|
|
|
|
supplied_identity = (
|
|
receipt.get("identity")
|
|
if isinstance(receipt.get("identity"), dict)
|
|
else receipt
|
|
)
|
|
if not isinstance(supplied_identity, dict):
|
|
return False
|
|
try:
|
|
parsed = parse_agent99_dispatch_identity(supplied_identity)
|
|
except ValueError:
|
|
return False
|
|
return parsed == identity and parsed.public_dict() == identity.public_dict()
|
|
|
|
|
|
def _safe_receipt_refs(receipt_refs: dict[str, Any] | None) -> dict[str, str]:
|
|
"""Keep only bounded durable identifiers, never raw evidence bodies."""
|
|
|
|
return sanitize_agent99_public_receipt_refs(receipt_refs)
|
|
|
|
|
|
class PostgresAgent99DispatchLedger:
|
|
"""PostgreSQL reservation and public-safe receipt store."""
|
|
|
|
async def reserve(
|
|
self,
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
payload: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
identity_payload = identity.public_dict()
|
|
input_payload = {
|
|
"schema_version": "agent99_controlled_dispatch_input_v1",
|
|
"identity": identity_payload,
|
|
"kind": str(payload.get("kind") or ""),
|
|
"suggested_mode": str(payload.get("suggestedMode") or "Status"),
|
|
"controlled_apply": bool(payload.get("controlledApply") is True),
|
|
"target_resource": str((payload.get("awoooi") or {}).get("targetResource") or ""),
|
|
"stores_secrets": False,
|
|
}
|
|
input_json = _stable_json(input_payload)
|
|
now = _utc_now_naive()
|
|
claim_token = str(uuid4())
|
|
stage_identity = {
|
|
"run_id": str(identity.run_id),
|
|
"trace_id": identity.trace_id,
|
|
"work_item_id": identity.work_item_id,
|
|
}
|
|
reservation_envelope = {
|
|
"schema_version": "agent99_controlled_dispatch_receipt_v1",
|
|
"status": "dispatch_reserved",
|
|
"identity": identity_payload,
|
|
"dispatch_accepted": False,
|
|
"controlled_apply_requested": bool(input_payload["controlled_apply"]),
|
|
"controlled_apply_authorized": False,
|
|
"runtime_execution_authorized": False,
|
|
"runtime_execution_attempted": False,
|
|
"post_verifier_passed": False,
|
|
"runtime_closure_verified": False,
|
|
"closure_complete": False,
|
|
"outbox": {
|
|
"status": "pending_delivery",
|
|
"durable_reservation": True,
|
|
"step_seq": 1,
|
|
**stage_identity,
|
|
},
|
|
"verifier": {
|
|
"status": "pending_dispatch",
|
|
"step_seq": 2,
|
|
**stage_identity,
|
|
},
|
|
"learning_writeback": {
|
|
"status": "pending_verifier",
|
|
"step_seq": 3,
|
|
**stage_identity,
|
|
},
|
|
}
|
|
|
|
try:
|
|
async with get_db_context(identity.project_id) as db:
|
|
await db.execute(
|
|
pg_insert(AwoooPRunState)
|
|
.values(
|
|
run_id=identity.run_id,
|
|
project_id=identity.project_id,
|
|
agent_id=AGENT99_DISPATCH_AGENT_ID,
|
|
state="pending",
|
|
attempt_count=0,
|
|
max_attempts=3,
|
|
trace_id=identity.trace_id,
|
|
trigger_type=AGENT99_DISPATCH_TRIGGER_TYPE,
|
|
trigger_ref=identity.trigger_ref,
|
|
is_shadow=False,
|
|
input_sha256=_sha256(input_json),
|
|
step_count=3,
|
|
error_detail=_stable_json(reservation_envelope),
|
|
timeout_at=(
|
|
now + timedelta(minutes=AGENT99_DISPATCH_TIMEOUT_MINUTES)
|
|
),
|
|
next_attempt_at=None,
|
|
)
|
|
.on_conflict_do_nothing(index_elements=[AwoooPRunState.run_id])
|
|
)
|
|
await db.execute(
|
|
pg_insert(AwoooPRunIdempotency)
|
|
.values(
|
|
project_id=identity.project_id,
|
|
channel_type=AGENT99_DISPATCH_CHANNEL,
|
|
provider_event_id=identity.idempotency_key,
|
|
run_id=identity.run_id,
|
|
)
|
|
.on_conflict_do_nothing(constraint="uix_run_idempotency_key")
|
|
)
|
|
claim = await db.execute(
|
|
update(AwoooPRunState)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
AwoooPRunState.attempt_count < AwoooPRunState.max_attempts,
|
|
# A running claim is delivery-unknown after a worker
|
|
# crash and is therefore reconciliation-only. Only a
|
|
# durable pending row whose not-before elapsed may be
|
|
# claimed for transport.
|
|
and_(
|
|
AwoooPRunState.state == "pending",
|
|
or_(
|
|
AwoooPRunState.next_attempt_at.is_(None),
|
|
AwoooPRunState.next_attempt_at <= now,
|
|
),
|
|
),
|
|
)
|
|
.values(
|
|
state="running",
|
|
# A fresh token fences every claim/reclaim. The stable
|
|
# run identity must never be used as a worker token.
|
|
worker_id=claim_token,
|
|
started_at=now,
|
|
heartbeat_at=now,
|
|
lease_until=(
|
|
now + timedelta(seconds=AGENT99_DISPATCH_LEASE_SECONDS)
|
|
),
|
|
next_attempt_at=None,
|
|
attempt_count=AwoooPRunState.attempt_count + 1,
|
|
)
|
|
.returning(AwoooPRunState.run_id)
|
|
)
|
|
claimed = claim.scalar_one_or_none() is not None
|
|
await db.execute(
|
|
pg_insert(AwoooPRunStepJournal)
|
|
.values([
|
|
{
|
|
"run_id": identity.run_id,
|
|
"project_id": identity.project_id,
|
|
"step_seq": 1,
|
|
"tool_name": "agent99.controlled_dispatch.outbox",
|
|
"input_hash": _sha256(input_json),
|
|
"compensation_json": reservation_envelope,
|
|
"result_status": "pending",
|
|
"was_blocked": False,
|
|
},
|
|
{
|
|
"run_id": identity.run_id,
|
|
"project_id": identity.project_id,
|
|
"step_seq": 2,
|
|
"tool_name": "agent99.independent_post_verifier",
|
|
"input_hash": _sha256(identity.trace_id),
|
|
"compensation_json": reservation_envelope["verifier"],
|
|
"result_status": "pending",
|
|
"was_blocked": False,
|
|
},
|
|
{
|
|
"run_id": identity.run_id,
|
|
"project_id": identity.project_id,
|
|
"step_seq": 3,
|
|
"tool_name": "agent99.learning_writeback",
|
|
"input_hash": _sha256(identity.work_item_id),
|
|
"compensation_json": reservation_envelope[
|
|
"learning_writeback"
|
|
],
|
|
"result_status": "pending",
|
|
"was_blocked": False,
|
|
},
|
|
])
|
|
.on_conflict_do_nothing(constraint="uix_run_step_seq")
|
|
)
|
|
state_result = await db.execute(
|
|
select(
|
|
AwoooPRunState.state,
|
|
AwoooPRunState.error_detail,
|
|
AwoooPRunState.next_attempt_at,
|
|
).where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
)
|
|
)
|
|
state_row = state_result.one_or_none()
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_dispatch_pg_reservation_failed_fail_closed",
|
|
incident_id=identity.incident_id,
|
|
run_id=str(identity.run_id),
|
|
error=str(exc),
|
|
)
|
|
return {
|
|
"status": "ledger_unavailable_fail_closed",
|
|
"claimed": False,
|
|
"identity": identity_payload,
|
|
"receipt": None,
|
|
}
|
|
|
|
existing = _parse_envelope(state_row.error_detail if state_row else None)
|
|
if claimed:
|
|
return {
|
|
"status": "reserved",
|
|
"claimed": True,
|
|
"identity": identity_payload,
|
|
"receipt": reservation_envelope,
|
|
"claim_token": claim_token,
|
|
}
|
|
next_attempt_at = getattr(state_row, "next_attempt_at", None)
|
|
retry_not_before = bool(next_attempt_at and next_attempt_at > now)
|
|
return {
|
|
"status": (
|
|
"retry_not_before"
|
|
if retry_not_before
|
|
else f"idempotent_{str(state_row.state if state_row else 'unknown')}"
|
|
),
|
|
"claimed": False,
|
|
"identity": identity_payload,
|
|
"receipt": existing,
|
|
"next_attempt_at": (
|
|
next_attempt_at.isoformat() if next_attempt_at else None
|
|
),
|
|
}
|
|
|
|
async def defer_claim_retryable(
|
|
self,
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
claim_token: str,
|
|
reason: str,
|
|
retry_after_seconds: int,
|
|
) -> dict[str, Any]:
|
|
"""Return an undispatched claim to pending without losing fencing.
|
|
|
|
A Redis lock collision or Redis outage occurs before transport. The run
|
|
therefore remains retryable; it must not be marked as a failed repair.
|
|
The claim token prevents an expired worker from changing a newer claim.
|
|
"""
|
|
|
|
now = _utc_now_naive()
|
|
safe_reason = str(reason or "dispatch_lock_unavailable")[:128]
|
|
retry_after = max(1, min(int(retry_after_seconds or 1), 3600))
|
|
next_attempt_at = now + timedelta(seconds=retry_after)
|
|
try:
|
|
async with get_db_context(identity.project_id) as db:
|
|
current_result = await db.execute(
|
|
select(
|
|
AwoooPRunState.state,
|
|
AwoooPRunState.worker_id,
|
|
AwoooPRunState.attempt_count,
|
|
AwoooPRunState.max_attempts,
|
|
AwoooPRunState.error_detail,
|
|
)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
)
|
|
.with_for_update()
|
|
)
|
|
current = current_result.one_or_none()
|
|
if (
|
|
current is None
|
|
or current.state != "running"
|
|
or str(current.worker_id or "") != str(claim_token or "")
|
|
):
|
|
return {
|
|
"status": "claim_fenced_no_write",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
envelope = _parse_envelope(current.error_detail) or {
|
|
"schema_version": "agent99_controlled_dispatch_receipt_v1",
|
|
"identity": identity.public_dict(),
|
|
}
|
|
attempts_exhausted = bool(
|
|
int(current.attempt_count or 0) >= int(current.max_attempts or 0)
|
|
)
|
|
envelope.update({
|
|
"status": "dispatch_lock_retryable",
|
|
"dispatch_accepted": False,
|
|
"inbox_triggered": False,
|
|
"dispatch_promoted": False,
|
|
"controlled_apply_authorized": False,
|
|
"runtime_execution_authorized": False,
|
|
"runtime_closure_verified": False,
|
|
"closure_complete": False,
|
|
"retry_policy": {
|
|
"safe_to_retry": True,
|
|
"retry_same_generation": not attempts_exhausted,
|
|
"requires_generation_advance": attempts_exhausted,
|
|
"reconcile_only": False,
|
|
"max_generation": AGENT99_MAX_EXECUTION_GENERATION,
|
|
"retry_after_seconds": retry_after,
|
|
"next_attempt_at": next_attempt_at.isoformat(),
|
|
"reason": safe_reason,
|
|
},
|
|
"outbox": {
|
|
"status": "lock_wait_retryable",
|
|
"durable_reservation": True,
|
|
"step_seq": 1,
|
|
**_stage_identity(identity),
|
|
},
|
|
})
|
|
envelope_json = _stable_json(envelope)
|
|
update_result = await db.execute(
|
|
update(AwoooPRunState)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
AwoooPRunState.state == "running",
|
|
AwoooPRunState.worker_id == claim_token,
|
|
)
|
|
.values(
|
|
state="pending",
|
|
worker_id=None,
|
|
lease_until=None,
|
|
next_attempt_at=next_attempt_at,
|
|
heartbeat_at=now,
|
|
error_detail=envelope_json,
|
|
)
|
|
.returning(AwoooPRunState.run_id)
|
|
)
|
|
persisted = update_result.scalar_one_or_none() is not None
|
|
if persisted:
|
|
await db.execute(
|
|
update(AwoooPRunStepJournal)
|
|
.where(
|
|
AwoooPRunStepJournal.run_id == identity.run_id,
|
|
AwoooPRunStepJournal.step_seq == 1,
|
|
)
|
|
.values(
|
|
compensation_json=envelope,
|
|
result_status="pending",
|
|
error_code=None,
|
|
was_blocked=True,
|
|
block_reason=safe_reason,
|
|
completed_at=None,
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_dispatch_retry_defer_failed_fail_closed",
|
|
incident_id=identity.incident_id,
|
|
run_id=str(identity.run_id),
|
|
error=str(exc),
|
|
)
|
|
return {
|
|
"status": "retry_defer_persistence_failed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
return {
|
|
**envelope,
|
|
"receipt_persisted": persisted,
|
|
"next_attempt_at": next_attempt_at.isoformat(),
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
async def complete(
|
|
self,
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
claim_token: str,
|
|
dispatch_receipt: dict[str, Any],
|
|
controlled_apply_authorized: bool | None = None,
|
|
) -> dict[str, Any]:
|
|
envelope = build_agent99_dispatch_receipt_envelope(
|
|
identity=identity,
|
|
dispatch_receipt=dispatch_receipt,
|
|
controlled_apply_authorized=controlled_apply_authorized,
|
|
)
|
|
accepted = bool(dispatch_receipt.get("accepted") is True)
|
|
inbox_triggered = bool(dispatch_receipt.get("inbox_triggered") is True)
|
|
dispatch_promoted = bool(accepted and inbox_triggered)
|
|
retry_safe = bool(
|
|
isinstance(envelope.get("retry_policy"), dict)
|
|
and envelope["retry_policy"].get("safe_to_retry") is True
|
|
)
|
|
now = _utc_now_naive()
|
|
next_attempt_at = (
|
|
now + timedelta(seconds=AGENT99_DISPATCH_RETRY_SECONDS)
|
|
if retry_safe
|
|
else None
|
|
)
|
|
if retry_safe:
|
|
envelope["retry_policy"].update({
|
|
"retry_after_seconds": AGENT99_DISPATCH_RETRY_SECONDS,
|
|
"next_attempt_at": next_attempt_at.isoformat(),
|
|
})
|
|
envelope_json = _stable_json(envelope)
|
|
try:
|
|
async with get_db_context(identity.project_id) as db:
|
|
result = await db.execute(
|
|
update(AwoooPRunState)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
AwoooPRunState.state.in_(["running", "waiting_tool"]),
|
|
AwoooPRunState.worker_id == claim_token,
|
|
)
|
|
.values(
|
|
# Delivery is not the terminal run state. A successful
|
|
# dispatch waits for the independent verifier and KM /
|
|
# PlayBook acknowledgements in steps 2 and 3.
|
|
state=(
|
|
"waiting_tool"
|
|
if dispatch_promoted or not retry_safe
|
|
else "failed"
|
|
),
|
|
output_sha256=_sha256(envelope_json),
|
|
error_code=(
|
|
None
|
|
if dispatch_promoted
|
|
else "E-AGENT99-DISPATCH-NOT-DELIVERED"
|
|
if retry_safe
|
|
else "E-AGENT99-DISPATCH-UNKNOWN"
|
|
),
|
|
error_detail=envelope_json,
|
|
completed_at=now if retry_safe else None,
|
|
heartbeat_at=now,
|
|
lease_until=None,
|
|
next_attempt_at=next_attempt_at,
|
|
worker_id=None,
|
|
)
|
|
.returning(AwoooPRunState.run_id)
|
|
)
|
|
finalized = result.scalar_one_or_none() is not None
|
|
if finalized:
|
|
await db.execute(
|
|
update(AwoooPRunStepJournal)
|
|
.where(
|
|
AwoooPRunStepJournal.run_id == identity.run_id,
|
|
AwoooPRunStepJournal.step_seq == 1,
|
|
)
|
|
.values(
|
|
output_hash=_sha256(envelope_json),
|
|
compensation_json=envelope,
|
|
result_status=(
|
|
"success"
|
|
if dispatch_promoted
|
|
else "failed"
|
|
if retry_safe
|
|
else "pending"
|
|
),
|
|
error_code=(
|
|
None
|
|
if dispatch_promoted
|
|
else "E-AGENT99-DISPATCH-NOT-DELIVERED"
|
|
if retry_safe
|
|
else None
|
|
),
|
|
was_blocked=not dispatch_promoted,
|
|
block_reason=(
|
|
None
|
|
if dispatch_promoted
|
|
else "dispatch_not_delivered"
|
|
if retry_safe
|
|
else "dispatch_delivery_reconcile_pending"
|
|
),
|
|
completed_at=now if dispatch_promoted or retry_safe else None,
|
|
)
|
|
)
|
|
if retry_safe:
|
|
await db.execute(
|
|
update(AwoooPRunStepJournal)
|
|
.where(
|
|
AwoooPRunStepJournal.run_id == identity.run_id,
|
|
AwoooPRunStepJournal.step_seq.in_([2, 3]),
|
|
AwoooPRunStepJournal.result_status == "pending",
|
|
)
|
|
.values(
|
|
result_status="failed",
|
|
error_code="E-AGENT99-DISPATCH-NOT-DELIVERED",
|
|
was_blocked=True,
|
|
block_reason="dispatch_not_accepted",
|
|
completed_at=now,
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_dispatch_pg_receipt_write_failed_fail_closed",
|
|
incident_id=identity.incident_id,
|
|
run_id=str(identity.run_id),
|
|
error=str(exc),
|
|
)
|
|
return {
|
|
**envelope,
|
|
"status": "receipt_persistence_failed_no_runtime_closure",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
return {
|
|
**envelope,
|
|
"receipt_persisted": finalized,
|
|
"claim_fenced": not finalized,
|
|
"status": (
|
|
str(envelope.get("status") or "unknown")
|
|
if finalized
|
|
else "claim_fenced_no_write"
|
|
),
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
async def mark_delivery_attempt_started(
|
|
self,
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
claim_token: str,
|
|
) -> dict[str, Any]:
|
|
"""Fence transport ambiguity before the first network/file write.
|
|
|
|
After this durable transition a worker crash is delivery-unknown. The
|
|
run stays in ``waiting_tool`` and only authenticated outcome readback
|
|
may advance it; an expired lease can never redispatch the mutation.
|
|
"""
|
|
|
|
now = _utc_now_naive()
|
|
try:
|
|
async with get_db_context(identity.project_id) as db:
|
|
current_result = await db.execute(
|
|
select(
|
|
AwoooPRunState.state,
|
|
AwoooPRunState.worker_id,
|
|
AwoooPRunState.error_detail,
|
|
)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
)
|
|
.with_for_update()
|
|
)
|
|
current = current_result.one_or_none()
|
|
if (
|
|
current is None
|
|
or current.state != "running"
|
|
or str(current.worker_id or "") != str(claim_token or "")
|
|
):
|
|
return {
|
|
"status": "claim_fenced_no_write",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
envelope = _parse_envelope(current.error_detail) or {
|
|
"schema_version": "agent99_controlled_dispatch_receipt_v1",
|
|
"identity": identity.public_dict(),
|
|
}
|
|
envelope.update({
|
|
"status": "dispatch_delivery_unknown_reconcile_only",
|
|
"dispatch_accepted": False,
|
|
"inbox_triggered": False,
|
|
"dispatch_promoted": False,
|
|
"controlled_apply_authorized": False,
|
|
"runtime_execution_authorized": False,
|
|
"runtime_closure_verified": False,
|
|
"closure_complete": False,
|
|
"retry_policy": {
|
|
"safe_to_retry": False,
|
|
"requires_generation_advance": False,
|
|
"reconcile_only": True,
|
|
"max_generation": AGENT99_MAX_EXECUTION_GENERATION,
|
|
},
|
|
"outbox": {
|
|
"status": "delivery_attempt_started_unknown",
|
|
"durable_reservation": True,
|
|
"step_seq": 1,
|
|
**_stage_identity(identity),
|
|
},
|
|
})
|
|
envelope_json = _stable_json(envelope)
|
|
updated = await db.execute(
|
|
update(AwoooPRunState)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
AwoooPRunState.state == "running",
|
|
AwoooPRunState.worker_id == claim_token,
|
|
)
|
|
.values(
|
|
state="waiting_tool",
|
|
error_code="E-AGENT99-DISPATCH-UNKNOWN",
|
|
error_detail=envelope_json,
|
|
heartbeat_at=now,
|
|
next_attempt_at=None,
|
|
)
|
|
.returning(AwoooPRunState.run_id)
|
|
)
|
|
persisted = updated.scalar_one_or_none() is not None
|
|
if persisted:
|
|
await db.execute(
|
|
update(AwoooPRunStepJournal)
|
|
.where(
|
|
AwoooPRunStepJournal.run_id == identity.run_id,
|
|
AwoooPRunStepJournal.step_seq == 1,
|
|
)
|
|
.values(
|
|
compensation_json=envelope["outbox"],
|
|
result_status="pending",
|
|
error_code=None,
|
|
was_blocked=True,
|
|
block_reason="dispatch_delivery_reconcile_pending",
|
|
completed_at=None,
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_delivery_attempt_fence_failed_fail_closed",
|
|
incident_id=identity.incident_id,
|
|
run_id=str(identity.run_id),
|
|
error=str(exc),
|
|
)
|
|
return {
|
|
"status": "delivery_attempt_fence_persistence_failed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
return {
|
|
**envelope,
|
|
"receipt_persisted": persisted,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
async def record_verifier(
|
|
self,
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
outcome_receipt: dict[str, Any],
|
|
evidence_refs: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Persist Agent99 outcome + independent verifier on the same run."""
|
|
|
|
if not validate_agent99_outcome_identity(identity, outcome_receipt):
|
|
return {
|
|
"status": "verifier_identity_mismatch_fail_closed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
outcome = (
|
|
outcome_receipt.get("outcome")
|
|
if isinstance(outcome_receipt.get("outcome"), dict)
|
|
else outcome_receipt
|
|
)
|
|
schema_version = str(
|
|
outcome.get("schemaVersion") or outcome.get("schema_version") or ""
|
|
)
|
|
if schema_version != "agent99_outcome_contract_v1":
|
|
return {
|
|
"status": "verifier_schema_invalid_fail_closed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
outcome_state = str(outcome.get("state") or "unknown")
|
|
verifier_passed = bool(outcome.get("verifierPassed") is True)
|
|
source_resolved = bool(outcome.get("sourceEventResolved") is True)
|
|
transport_ok = bool(outcome.get("transportOk") is True)
|
|
safe_refs = _safe_receipt_refs(evidence_refs)
|
|
required_evidence_refs = set(AGENT99_REQUIRED_VERIFIER_EVIDENCE_REFS)
|
|
mode = str(outcome_receipt.get("mode") or "")
|
|
if mode == "BackupCheck" or "backup_health" in identity.route_id:
|
|
required_evidence_refs.update(
|
|
AGENT99_BACKUP_REQUIRED_VERIFIER_EVIDENCE_REFS
|
|
)
|
|
missing_evidence_refs = sorted(required_evidence_refs - set(safe_refs))
|
|
if missing_evidence_refs:
|
|
return {
|
|
"status": "verifier_evidence_missing_fail_closed",
|
|
"missing_evidence_refs": missing_evidence_refs,
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
passed = bool(
|
|
outcome_state == "resolved"
|
|
and transport_ok
|
|
and verifier_passed
|
|
and source_resolved
|
|
)
|
|
execution_attempted = transport_ok
|
|
now = _utc_now_naive()
|
|
|
|
try:
|
|
async with get_db_context(identity.project_id) as db:
|
|
current_result = await db.execute(
|
|
select(AwoooPRunState.state, AwoooPRunState.error_detail)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
)
|
|
.with_for_update()
|
|
)
|
|
current = current_result.one_or_none()
|
|
envelope = _parse_envelope(
|
|
current.error_detail if current is not None else None
|
|
)
|
|
if (
|
|
current is None
|
|
or current.state not in {"pending", "running", "waiting_tool"}
|
|
or not isinstance(envelope, dict)
|
|
or not _stage_identity_matches(
|
|
identity,
|
|
envelope.get("identity")
|
|
if isinstance(envelope.get("identity"), dict)
|
|
else {},
|
|
)
|
|
):
|
|
return {
|
|
"status": "verifier_run_not_ready_fail_closed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
if (
|
|
current.state in {"pending", "running"}
|
|
or envelope.get("dispatch_promoted") is not True
|
|
):
|
|
# A signed/allowlisted Agent99 outcome is the authoritative
|
|
# reconciliation receipt when transport acceptance was
|
|
# observed but the API could not persist its response.
|
|
envelope.update({
|
|
"dispatch_accepted": True,
|
|
"inbox_triggered": True,
|
|
"dispatch_promoted": True,
|
|
"controlled_apply_authorized": bool(
|
|
outcome_receipt.get("controlledApply") is True
|
|
),
|
|
"dispatch_receipt": {
|
|
"schema_version": "agent99_sre_dispatch_receipt_v1",
|
|
"status": "outcome_readback_proves_delivery",
|
|
"transport": "agent99_relay_readback",
|
|
"alert_id": str(identity.run_id),
|
|
"kind": str(outcome_receipt.get("mode") or ""),
|
|
"accepted": True,
|
|
"inbox_triggered": True,
|
|
"delivery_certainty": "delivered",
|
|
"stores_raw_response": False,
|
|
},
|
|
"outbox": {
|
|
"status": "delivered_reconciled_from_outcome",
|
|
"durable_reservation": True,
|
|
"step_seq": 1,
|
|
**_stage_identity(identity),
|
|
},
|
|
"retry_policy": {
|
|
"safe_to_retry": False,
|
|
"requires_generation_advance": False,
|
|
"reconcile_only": False,
|
|
"max_generation": AGENT99_MAX_EXECUTION_GENERATION,
|
|
},
|
|
})
|
|
|
|
verifier_receipt = {
|
|
"schema_version": "agent99_independent_verifier_receipt_v1",
|
|
"status": "success" if passed else "failed",
|
|
**_stage_identity(identity),
|
|
"outcome_schema_version": schema_version,
|
|
"outcome_state": outcome_state,
|
|
"transport_ok": transport_ok,
|
|
"verifier_passed": verifier_passed,
|
|
"source_event_resolved": source_resolved,
|
|
"verified_at": str(
|
|
outcome.get("verifiedAt")
|
|
or outcome.get("verified_at")
|
|
or ""
|
|
)[:64],
|
|
"evidence_refs": safe_refs,
|
|
"stores_raw_evidence": False,
|
|
}
|
|
envelope.update({
|
|
"status": (
|
|
"verifier_passed_learning_writeback_pending"
|
|
if passed
|
|
else "verifier_failed_no_runtime_closure"
|
|
),
|
|
"runtime_execution_authorized": bool(
|
|
envelope.get("controlled_apply_authorized") is True
|
|
),
|
|
"runtime_execution_attempted": execution_attempted,
|
|
"post_verifier_passed": passed,
|
|
"runtime_closure_verified": False,
|
|
"closure_complete": False,
|
|
"verifier": verifier_receipt,
|
|
"learning_writeback": {
|
|
**_stage_identity(identity),
|
|
"step_seq": 3,
|
|
"status": (
|
|
"pending_writeback"
|
|
if passed
|
|
else "blocked_verifier_failed"
|
|
),
|
|
"required_receipts": [
|
|
"incident_closure_receipt",
|
|
"telegram_lifecycle_receipt",
|
|
"km_writeback_ack",
|
|
"playbook_trust_writeback_ack",
|
|
],
|
|
},
|
|
})
|
|
envelope_json = _stable_json(envelope)
|
|
update_result = await db.execute(
|
|
update(AwoooPRunState)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
AwoooPRunState.state.in_([
|
|
"pending",
|
|
"running",
|
|
"waiting_tool",
|
|
]),
|
|
)
|
|
.values(
|
|
state="waiting_tool" if passed else "failed",
|
|
output_sha256=_sha256(envelope_json),
|
|
error_code=None if passed else "E-AGENT99-VERIFY",
|
|
error_detail=envelope_json,
|
|
heartbeat_at=now,
|
|
completed_at=None if passed else now,
|
|
lease_until=None,
|
|
next_attempt_at=None,
|
|
worker_id=None,
|
|
)
|
|
.returning(AwoooPRunState.run_id)
|
|
)
|
|
persisted = update_result.scalar_one_or_none() is not None
|
|
if persisted:
|
|
if current.state == "running":
|
|
await db.execute(
|
|
update(AwoooPRunStepJournal)
|
|
.where(
|
|
AwoooPRunStepJournal.run_id == identity.run_id,
|
|
AwoooPRunStepJournal.step_seq == 1,
|
|
)
|
|
.values(
|
|
output_hash=_sha256(envelope_json),
|
|
compensation_json=envelope["outbox"],
|
|
result_status="success",
|
|
error_code=None,
|
|
was_blocked=False,
|
|
block_reason=None,
|
|
completed_at=now,
|
|
)
|
|
)
|
|
await db.execute(
|
|
update(AwoooPRunStepJournal)
|
|
.where(
|
|
AwoooPRunStepJournal.run_id == identity.run_id,
|
|
AwoooPRunStepJournal.step_seq == 2,
|
|
)
|
|
.values(
|
|
output_hash=_sha256(_stable_json(verifier_receipt)),
|
|
compensation_json=verifier_receipt,
|
|
result_status="success" if passed else "failed",
|
|
error_code=None if passed else "E-AGENT99-VERIFY",
|
|
was_blocked=not passed,
|
|
block_reason=(
|
|
None if passed else "independent_verifier_failed"
|
|
),
|
|
completed_at=now,
|
|
)
|
|
)
|
|
if not passed:
|
|
await db.execute(
|
|
update(AwoooPRunStepJournal)
|
|
.where(
|
|
AwoooPRunStepJournal.run_id == identity.run_id,
|
|
AwoooPRunStepJournal.step_seq == 3,
|
|
AwoooPRunStepJournal.result_status == "pending",
|
|
)
|
|
.values(
|
|
result_status="failed",
|
|
error_code="E-AGENT99-VERIFY",
|
|
was_blocked=True,
|
|
block_reason="independent_verifier_failed",
|
|
completed_at=now,
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_verifier_receipt_write_failed_fail_closed",
|
|
incident_id=identity.incident_id,
|
|
run_id=str(identity.run_id),
|
|
error=str(exc),
|
|
)
|
|
return {
|
|
"status": "verifier_receipt_persistence_failed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
return {
|
|
**envelope,
|
|
"receipt_persisted": persisted,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
async def record_external_no_write_reconciliation(
|
|
self,
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
outcome_receipt: dict[str, Any],
|
|
source_receipt: dict[str, Any],
|
|
evidence_refs: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Terminalize an externally recovered run without automation credit.
|
|
|
|
This is intentionally separate from ``record_verifier`` and
|
|
``record_learning_writeback``. A no-write Status proves current
|
|
posture only; it never proves that the original Recover execution
|
|
succeeded and must not resolve the incident or increase PlayBook
|
|
trust.
|
|
"""
|
|
|
|
if not validate_agent99_outcome_identity(identity, outcome_receipt):
|
|
return {
|
|
"status": "external_recovery_identity_mismatch_fail_closed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
outcome = (
|
|
outcome_receipt.get("outcome")
|
|
if isinstance(outcome_receipt.get("outcome"), dict)
|
|
else {}
|
|
)
|
|
expected_top_level = {
|
|
"automationRunId": str(identity.run_id),
|
|
"traceId": identity.trace_id,
|
|
"workItemId": identity.work_item_id,
|
|
"projectId": identity.project_id,
|
|
"incidentId": identity.incident_id,
|
|
"routeId": identity.route_id,
|
|
"executionGeneration": identity.execution_generation,
|
|
"approvalId": identity.approval_id,
|
|
}
|
|
if any(
|
|
str(outcome_receipt.get(key) or "") != expected
|
|
for key, expected in expected_top_level.items()
|
|
):
|
|
return {
|
|
"status": "external_recovery_top_level_identity_fail_closed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
source_ref = normalize_agent99_public_receipt_ref(
|
|
source_receipt.get("receipt_ref")
|
|
)
|
|
source_values = (
|
|
source_receipt.get("values")
|
|
if isinstance(source_receipt.get("values"), dict)
|
|
else {}
|
|
)
|
|
source_result = str(source_receipt.get("source_result") or "")
|
|
source_counts = {
|
|
key: _agent99_exact_nonnegative_int(source_values.get(key))
|
|
for key in (
|
|
"monitor_up",
|
|
"pass_gates",
|
|
"warn_gates",
|
|
"blocked_gates",
|
|
"green_result",
|
|
"degraded_result",
|
|
"blocked_result",
|
|
"check_failed_result",
|
|
"firing_blocking_alerts",
|
|
)
|
|
}
|
|
source_age = source_receipt.get("age_seconds")
|
|
source_age_exact = bool(
|
|
not isinstance(source_age, bool)
|
|
and isinstance(source_age, int | float)
|
|
and float(source_age) >= -60
|
|
and float(source_age) <= 900
|
|
)
|
|
source_observed_at = _agent99_exact_nonnegative_int(
|
|
source_receipt.get("observed_at_epoch")
|
|
)
|
|
expected_evidence_id = (
|
|
f"same-run-status-{identity.run_id}-"
|
|
f"{identity.public_dict()['canonical_digest'][:12]}"
|
|
)
|
|
exact_outcome = bool(
|
|
outcome_receipt.get("schemaVersion")
|
|
== "agent99_outcome_readback_v1"
|
|
and outcome_receipt.get("sameRunContractFieldsComplete") is True
|
|
and outcome_receipt.get("mode") == "Status"
|
|
and outcome_receipt.get("controlledApply") is False
|
|
and outcome_receipt.get("reconcileOnly") is True
|
|
and outcome_receipt.get("runtimeWritePerformed") is False
|
|
and outcome_receipt.get("evidenceReceiptId")
|
|
== expected_evidence_id
|
|
and outcome.get("schemaVersion") == "agent99_outcome_contract_v1"
|
|
and outcome.get("state") == "resolved"
|
|
and outcome.get("resolved") is True
|
|
and outcome.get("transportOk") is True
|
|
and outcome.get("verifierPassed") is True
|
|
and outcome.get("verifierName")
|
|
== "agent99-status-same-run-no-write-post-condition-v1"
|
|
and outcome.get("sourceEventResolved") is True
|
|
and source_ref is not None
|
|
and outcome.get("sourceEventEvidence") == source_ref
|
|
)
|
|
source_exact = bool(
|
|
source_receipt.get("schema_version")
|
|
== "agent99_cold_start_source_resolution_v1"
|
|
and source_receipt.get("resolved") is True
|
|
and source_result in {"green", "degraded"}
|
|
and source_receipt.get("degraded_provenance")
|
|
is (source_result == "degraded")
|
|
and source_age_exact
|
|
and source_observed_at is not None
|
|
and source_observed_at > 0
|
|
and source_counts["monitor_up"] == 1
|
|
and source_counts["pass_gates"] is not None
|
|
and source_counts["pass_gates"] > 0
|
|
and source_counts["warn_gates"] is not None
|
|
and source_counts["blocked_gates"] == 0
|
|
and source_counts["blocked_result"] == 0
|
|
and source_counts["check_failed_result"] == 0
|
|
and source_counts["firing_blocking_alerts"] == 0
|
|
and (
|
|
(
|
|
source_result == "degraded"
|
|
and source_counts["warn_gates"] > 0
|
|
and source_counts["green_result"] == 0
|
|
and source_counts["degraded_result"] == 1
|
|
)
|
|
or (
|
|
source_result == "green"
|
|
and source_counts["warn_gates"] == 0
|
|
and source_counts["green_result"] == 1
|
|
and source_counts["degraded_result"] == 0
|
|
)
|
|
)
|
|
)
|
|
safe_refs = _safe_receipt_refs(evidence_refs)
|
|
missing_refs = sorted(
|
|
set(AGENT99_REQUIRED_VERIFIER_EVIDENCE_REFS) - set(safe_refs)
|
|
)
|
|
if not exact_outcome or not source_exact or missing_refs:
|
|
return {
|
|
"status": "external_recovery_contract_invalid_fail_closed",
|
|
"missing_evidence_refs": missing_refs,
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
if safe_refs.get("source_event_evidence_ref") != source_ref:
|
|
return {
|
|
"status": "external_recovery_source_receipt_mismatch_fail_closed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
now = _utc_now_naive()
|
|
terminal_receipt = {
|
|
"schema_version": (
|
|
"agent99_external_recovery_no_write_reconciliation_v1"
|
|
),
|
|
"status": "external_recovery_no_write_reconciled",
|
|
**_stage_identity(identity),
|
|
"mode": "Status",
|
|
"verifier_name": (
|
|
"agent99-status-same-run-no-write-post-condition-v1"
|
|
),
|
|
"evidence_receipt_id": expected_evidence_id,
|
|
"source": {
|
|
"receipt_ref": source_ref,
|
|
"result": source_result,
|
|
"degraded_provenance": bool(
|
|
source_receipt.get("degraded_provenance")
|
|
),
|
|
"pass_gates": source_counts["pass_gates"],
|
|
"warn_gates": source_counts["warn_gates"],
|
|
"blocked_gates": 0,
|
|
"age_seconds": source_age,
|
|
"observed_at_epoch": source_observed_at,
|
|
},
|
|
"transport_ok": True,
|
|
"status_verifier_passed": True,
|
|
"runtime_write_performed": False,
|
|
"recover_execution_success": False,
|
|
"incident_resolution_authorized": False,
|
|
"learning_writeback_authorized": False,
|
|
"telegram_authorized": False,
|
|
"evidence_refs": safe_refs,
|
|
"stores_raw_evidence": False,
|
|
}
|
|
|
|
try:
|
|
async with get_db_context(identity.project_id) as db:
|
|
current_result = await db.execute(
|
|
select(AwoooPRunState.state, AwoooPRunState.error_detail)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
)
|
|
.with_for_update()
|
|
)
|
|
current = current_result.one_or_none()
|
|
envelope = _parse_envelope(
|
|
current.error_detail if current is not None else None
|
|
)
|
|
if (
|
|
current is not None
|
|
and current.state == "cancelled"
|
|
and isinstance(envelope, dict)
|
|
and envelope.get("external_no_write_reconciliation_terminal")
|
|
is True
|
|
and _stage_identity_matches(
|
|
identity,
|
|
envelope.get("identity")
|
|
if isinstance(envelope.get("identity"), dict)
|
|
else {},
|
|
)
|
|
):
|
|
return {
|
|
**envelope,
|
|
"status": (
|
|
"external_recovery_no_write_terminal_idempotent"
|
|
),
|
|
"receipt_persisted": True,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
if (
|
|
current is None
|
|
or current.state
|
|
not in {"pending", "running", "waiting_tool"}
|
|
or not isinstance(envelope, dict)
|
|
or not _stage_identity_matches(
|
|
identity,
|
|
envelope.get("identity")
|
|
if isinstance(envelope.get("identity"), dict)
|
|
else {},
|
|
)
|
|
):
|
|
return {
|
|
"status": (
|
|
"external_recovery_run_not_ready_fail_closed"
|
|
),
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
prior_runtime_execution_attempted = bool(
|
|
envelope.get("runtime_execution_attempted") is True
|
|
)
|
|
envelope.update({
|
|
"status": "external_recovery_no_write_reconciled_terminal",
|
|
"run_terminal_state": "cancelled",
|
|
"controlled_apply_authorized": False,
|
|
"runtime_execution_authorized": False,
|
|
"runtime_execution_attempted": (
|
|
prior_runtime_execution_attempted
|
|
),
|
|
"same_run_status_runtime_execution_attempted": False,
|
|
"runtime_write_performed": False,
|
|
"automation_execution_success": False,
|
|
"post_verifier_passed": False,
|
|
"external_status_verifier_passed": True,
|
|
"external_recovery_source_resolved": True,
|
|
"runtime_closure_verified": False,
|
|
"closure_complete": False,
|
|
"incident_resolution_authorized": False,
|
|
"external_no_write_reconciliation_terminal": True,
|
|
"external_recovery_reconciliation": terminal_receipt,
|
|
"learning_writeback": {
|
|
**_stage_identity(identity),
|
|
"step_seq": 3,
|
|
"status": (
|
|
"not_applicable_external_recovery_no_write"
|
|
),
|
|
"required_receipts": [],
|
|
"receipt_refs": {},
|
|
"stores_raw_evidence": False,
|
|
},
|
|
})
|
|
envelope_json = _stable_json(envelope)
|
|
updated = await db.execute(
|
|
update(AwoooPRunState)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
AwoooPRunState.state.in_([
|
|
"pending",
|
|
"running",
|
|
"waiting_tool",
|
|
]),
|
|
)
|
|
.values(
|
|
state="cancelled",
|
|
output_sha256=_sha256(envelope_json),
|
|
error_code=None,
|
|
error_detail=envelope_json,
|
|
heartbeat_at=now,
|
|
completed_at=now,
|
|
lease_until=None,
|
|
next_attempt_at=None,
|
|
worker_id=None,
|
|
)
|
|
.returning(AwoooPRunState.run_id)
|
|
)
|
|
persisted = updated.scalar_one_or_none() is not None
|
|
if persisted:
|
|
journal_states = {
|
|
1: (
|
|
"compensated",
|
|
True,
|
|
"external_recovery_superseded_runtime_execution",
|
|
),
|
|
2: ("success", False, None),
|
|
3: (
|
|
"compensated",
|
|
True,
|
|
"learning_not_applicable_external_recovery_no_write",
|
|
),
|
|
}
|
|
for step_seq, (
|
|
result_status,
|
|
was_blocked,
|
|
block_reason,
|
|
) in journal_states.items():
|
|
await db.execute(
|
|
update(AwoooPRunStepJournal)
|
|
.where(
|
|
AwoooPRunStepJournal.run_id == identity.run_id,
|
|
AwoooPRunStepJournal.step_seq == step_seq,
|
|
)
|
|
.values(
|
|
output_hash=_sha256(
|
|
_stable_json(terminal_receipt)
|
|
),
|
|
compensation_json=terminal_receipt,
|
|
result_status=result_status,
|
|
error_code=None,
|
|
was_blocked=was_blocked,
|
|
block_reason=block_reason,
|
|
completed_at=now,
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_external_no_write_reconciliation_failed",
|
|
incident_id=identity.incident_id,
|
|
run_id=str(identity.run_id),
|
|
error=str(exc),
|
|
)
|
|
return {
|
|
"status": "external_recovery_persistence_failed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
return {
|
|
**envelope,
|
|
"receipt_persisted": persisted,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
async def record_learning_writeback(
|
|
self,
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
receipt_refs: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""Atomically close the incident, operation log, and same-run ledger.
|
|
|
|
KM, PlayBook, DR, and Telegram each provide a durable idempotent
|
|
acknowledgement first. The incident terminal state, immutable closure
|
|
event, run state, and step-3 receipt then commit in one PostgreSQL
|
|
transaction. Redis is only a post-commit projection.
|
|
"""
|
|
|
|
required_refs = {
|
|
"telegram_lifecycle_receipt_id",
|
|
"km_writeback_ack_id",
|
|
"playbook_trust_writeback_ack_id",
|
|
}
|
|
if "backup_health" in identity.route_id:
|
|
required_refs.add("dr_scorecard_writeback_ack_id")
|
|
supplied_refs = sanitize_agent99_public_receipt_refs(
|
|
receipt_refs,
|
|
allowed_keys=(
|
|
AGENT99_LEARNING_RECEIPT_REFS
|
|
- {"incident_closure_receipt_id"}
|
|
),
|
|
)
|
|
supplied_missing = sorted(required_refs - set(supplied_refs))
|
|
if supplied_refs and supplied_missing:
|
|
return {
|
|
"status": "learning_writeback_incomplete_fail_closed",
|
|
"missing_receipts": supplied_missing,
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
now = _utc_now_naive()
|
|
incident_now = datetime.now(UTC)
|
|
closure_event_id = str(uuid5(
|
|
NAMESPACE_URL,
|
|
f"agent99-terminal-bundle:{identity.project_id}:{identity.run_id}",
|
|
))
|
|
try:
|
|
async with get_db_context(identity.project_id) as db:
|
|
current_result = await db.execute(
|
|
select(AwoooPRunState.state, AwoooPRunState.error_detail)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
)
|
|
.with_for_update()
|
|
)
|
|
current = current_result.one_or_none()
|
|
envelope = _parse_envelope(
|
|
current.error_detail if current is not None else None
|
|
)
|
|
if (
|
|
current is not None
|
|
and current.state == "completed"
|
|
and isinstance(envelope, dict)
|
|
and envelope.get("runtime_closure_verified") is True
|
|
and envelope.get("closure_complete") is True
|
|
and _stage_identity_matches(
|
|
identity,
|
|
envelope.get("identity")
|
|
if isinstance(envelope.get("identity"), dict)
|
|
else {},
|
|
)
|
|
):
|
|
return {
|
|
**envelope,
|
|
"status": "closed_verified_learning_written_idempotent",
|
|
"receipt_persisted": True,
|
|
"postgres_terminal_bundle_committed": True,
|
|
"redis_projection_pending": True,
|
|
"runtime_closure_verified": True,
|
|
}
|
|
if (
|
|
current is None
|
|
or current.state != "waiting_tool"
|
|
or not isinstance(envelope, dict)
|
|
or envelope.get("post_verifier_passed") is not True
|
|
or not _stage_identity_matches(
|
|
identity,
|
|
envelope.get("identity")
|
|
if isinstance(envelope.get("identity"), dict)
|
|
else {},
|
|
)
|
|
):
|
|
return {
|
|
"status": "learning_writeback_run_not_ready_fail_closed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
learning_checkpoint = (
|
|
envelope.get("learning_writeback")
|
|
if isinstance(envelope.get("learning_writeback"), dict)
|
|
else {}
|
|
)
|
|
checkpoint_refs = sanitize_agent99_public_receipt_refs(
|
|
learning_checkpoint.get("receipt_refs")
|
|
if isinstance(learning_checkpoint.get("receipt_refs"), dict)
|
|
else {},
|
|
allowed_keys=(
|
|
AGENT99_LEARNING_RECEIPT_REFS
|
|
- {"incident_closure_receipt_id"}
|
|
),
|
|
)
|
|
missing = sorted(required_refs - set(checkpoint_refs))
|
|
if missing:
|
|
return {
|
|
"status": (
|
|
"learning_writeback_checkpoint_incomplete_fail_closed"
|
|
),
|
|
"missing_receipts": missing,
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
if supplied_refs and supplied_refs != checkpoint_refs:
|
|
return {
|
|
"status": (
|
|
"learning_writeback_checkpoint_mismatch_fail_closed"
|
|
),
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
# Only the refs already committed by record_learning_asset_acks
|
|
# are allowed into the terminal bundle. Caller values are a
|
|
# consistency check, never the source of truth.
|
|
safe_refs = dict(checkpoint_refs)
|
|
safe_refs["incident_closure_receipt_id"] = (
|
|
f"alert_operation_log:{closure_event_id}"
|
|
)
|
|
|
|
incident_result = await db.execute(
|
|
select(IncidentRecord)
|
|
.where(
|
|
IncidentRecord.incident_id == identity.incident_id,
|
|
IncidentRecord.project_id == identity.project_id,
|
|
)
|
|
.with_for_update()
|
|
)
|
|
incident = incident_result.scalar_one_or_none()
|
|
if incident is None:
|
|
return {
|
|
"status": "incident_terminal_bundle_target_missing",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
incident_outcome = dict(incident.outcome or {})
|
|
incident_outcome.update({
|
|
"execution_success": True,
|
|
"effectiveness_score": max(
|
|
4,
|
|
int(incident_outcome.get("effectiveness_score") or 0),
|
|
),
|
|
"learning_notes": (
|
|
f"Agent99 same-run verifier passed: {identity.run_id}"
|
|
),
|
|
"should_remember": True,
|
|
"automation_run_id": str(identity.run_id),
|
|
"trace_id": identity.trace_id,
|
|
"work_item_id": identity.work_item_id,
|
|
})
|
|
|
|
learning_receipt = {
|
|
"schema_version": "agent99_learning_writeback_receipt_v1",
|
|
"status": "success",
|
|
"step_seq": 3,
|
|
**_stage_identity(identity),
|
|
"receipt_refs": safe_refs,
|
|
"stores_raw_evidence": False,
|
|
}
|
|
envelope.update({
|
|
"status": "closed_verified_learning_written",
|
|
"runtime_closure_verified": True,
|
|
"closure_complete": True,
|
|
"learning_writeback": learning_receipt,
|
|
})
|
|
envelope_json = _stable_json(envelope)
|
|
await db.execute(
|
|
pg_insert(AlertOperationLog)
|
|
.values(
|
|
id=closure_event_id,
|
|
incident_id=identity.incident_id,
|
|
approval_id=identity.approval_id or None,
|
|
event_type="RESOLVED",
|
|
actor="agent99_controlled_dispatch_reconciler",
|
|
action_detail=(
|
|
"Agent99 verifier and durable learning receipts closed"
|
|
),
|
|
success=True,
|
|
context={
|
|
"automation_run_id": str(identity.run_id),
|
|
"trace_id": identity.trace_id,
|
|
"work_item_id": identity.work_item_id,
|
|
"runtime_closure_verified": True,
|
|
"receipt_refs": safe_refs,
|
|
},
|
|
created_at=incident_now,
|
|
)
|
|
.on_conflict_do_nothing(
|
|
index_elements=[AlertOperationLog.id]
|
|
)
|
|
)
|
|
incident_update = await db.execute(
|
|
update(IncidentRecord)
|
|
.where(
|
|
IncidentRecord.incident_id == identity.incident_id,
|
|
IncidentRecord.project_id == identity.project_id,
|
|
)
|
|
.values(
|
|
status=IncidentStatus.RESOLVED,
|
|
outcome=incident_outcome,
|
|
resolved_at=incident_now,
|
|
updated_at=incident_now,
|
|
)
|
|
.returning(IncidentRecord.incident_id)
|
|
)
|
|
if incident_update.scalar_one_or_none() is None:
|
|
raise RuntimeError("agent99_incident_terminal_update_missing")
|
|
update_result = await db.execute(
|
|
update(AwoooPRunState)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
AwoooPRunState.state == "waiting_tool",
|
|
)
|
|
.values(
|
|
state="completed",
|
|
output_sha256=_sha256(envelope_json),
|
|
error_code=None,
|
|
error_detail=envelope_json,
|
|
heartbeat_at=now,
|
|
completed_at=now,
|
|
lease_until=None,
|
|
next_attempt_at=None,
|
|
worker_id=None,
|
|
)
|
|
.returning(AwoooPRunState.run_id)
|
|
)
|
|
persisted = update_result.scalar_one_or_none() is not None
|
|
if not persisted:
|
|
raise RuntimeError("agent99_terminal_run_update_fenced")
|
|
terminal_step = await db.execute(
|
|
update(AwoooPRunStepJournal)
|
|
.where(
|
|
AwoooPRunStepJournal.run_id == identity.run_id,
|
|
AwoooPRunStepJournal.step_seq == 3,
|
|
)
|
|
.values(
|
|
output_hash=_sha256(_stable_json(learning_receipt)),
|
|
compensation_json=learning_receipt,
|
|
result_status="success",
|
|
error_code=None,
|
|
was_blocked=False,
|
|
block_reason=None,
|
|
completed_at=now,
|
|
)
|
|
.returning(AwoooPRunStepJournal.step_id)
|
|
)
|
|
if terminal_step.scalar_one_or_none() is None:
|
|
raise RuntimeError("agent99_terminal_step_update_missing")
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_learning_writeback_failed_fail_closed",
|
|
incident_id=identity.incident_id,
|
|
run_id=str(identity.run_id),
|
|
error=str(exc),
|
|
)
|
|
return {
|
|
"status": "learning_writeback_persistence_failed",
|
|
"receipt_persisted": False,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
return {
|
|
**envelope,
|
|
"receipt_persisted": persisted,
|
|
"postgres_terminal_bundle_committed": persisted,
|
|
"redis_projection_pending": persisted,
|
|
"runtime_closure_verified": bool(
|
|
persisted and envelope.get("runtime_closure_verified") is True
|
|
),
|
|
}
|
|
|
|
async def record_learning_asset_acks(
|
|
self,
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
receipt_refs: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Checkpoint per-run learning side effects before terminal closure.
|
|
|
|
Reconciliation may crash after KM, PlayBook, Telegram, or DR succeeds.
|
|
Persisting each public acknowledgement on the same run lets the next
|
|
tick resume at the first missing asset instead of replaying completed
|
|
side effects. The asset writers retain their own deterministic keys as
|
|
the final crash fence between side effect and this checkpoint.
|
|
"""
|
|
|
|
supplied = sanitize_agent99_public_receipt_refs(
|
|
receipt_refs,
|
|
allowed_keys=(
|
|
AGENT99_LEARNING_RECEIPT_REFS
|
|
- {"incident_closure_receipt_id"}
|
|
),
|
|
)
|
|
now = _utc_now_naive()
|
|
try:
|
|
async with get_db_context(identity.project_id) as db:
|
|
current_result = await db.execute(
|
|
select(AwoooPRunState.state, AwoooPRunState.error_detail)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
)
|
|
.with_for_update()
|
|
)
|
|
current = current_result.one_or_none()
|
|
envelope = _parse_envelope(
|
|
current.error_detail if current is not None else None
|
|
)
|
|
if (
|
|
current is None
|
|
or current.state not in {"waiting_tool", "completed"}
|
|
or not isinstance(envelope, dict)
|
|
or envelope.get("post_verifier_passed") is not True
|
|
or not _stage_identity_matches(
|
|
identity,
|
|
envelope.get("identity")
|
|
if isinstance(envelope.get("identity"), dict)
|
|
else {},
|
|
)
|
|
):
|
|
return {
|
|
"status": "learning_asset_ack_run_not_ready_fail_closed",
|
|
"receipt_persisted": False,
|
|
"learning_receipt_refs": {},
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
learning = (
|
|
dict(envelope.get("learning_writeback"))
|
|
if isinstance(envelope.get("learning_writeback"), dict)
|
|
else {}
|
|
)
|
|
existing = sanitize_agent99_public_receipt_refs(
|
|
learning.get("receipt_refs")
|
|
if isinstance(learning.get("receipt_refs"), dict)
|
|
else {},
|
|
allowed_keys=AGENT99_LEARNING_RECEIPT_REFS,
|
|
)
|
|
merged = {**existing, **supplied}
|
|
if current.state == "completed":
|
|
return {
|
|
**envelope,
|
|
"status": "learning_asset_acks_terminal_readback",
|
|
"receipt_persisted": True,
|
|
"learning_receipt_refs": merged,
|
|
"runtime_closure_verified": bool(
|
|
envelope.get("runtime_closure_verified") is True
|
|
),
|
|
}
|
|
|
|
learning.update({
|
|
"status": "pending_writeback",
|
|
"step_seq": 3,
|
|
**_stage_identity(identity),
|
|
"receipt_refs": merged,
|
|
"acknowledged_receipts": sorted(merged),
|
|
"stores_raw_evidence": False,
|
|
})
|
|
envelope["learning_writeback"] = learning
|
|
envelope_json = _stable_json(envelope)
|
|
updated = await db.execute(
|
|
update(AwoooPRunState)
|
|
.where(
|
|
AwoooPRunState.run_id == identity.run_id,
|
|
AwoooPRunState.project_id == identity.project_id,
|
|
AwoooPRunState.state == "waiting_tool",
|
|
)
|
|
.values(
|
|
error_detail=envelope_json,
|
|
heartbeat_at=now,
|
|
)
|
|
.returning(AwoooPRunState.run_id)
|
|
)
|
|
persisted = updated.scalar_one_or_none() is not None
|
|
if persisted:
|
|
await db.execute(
|
|
update(AwoooPRunStepJournal)
|
|
.where(
|
|
AwoooPRunStepJournal.run_id == identity.run_id,
|
|
AwoooPRunStepJournal.step_seq == 3,
|
|
)
|
|
.values(
|
|
compensation_json=learning,
|
|
result_status="pending",
|
|
error_code=None,
|
|
was_blocked=False,
|
|
block_reason=None,
|
|
completed_at=None,
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_learning_asset_ack_failed_fail_closed",
|
|
incident_id=identity.incident_id,
|
|
run_id=str(identity.run_id),
|
|
error=str(exc),
|
|
)
|
|
return {
|
|
"status": "learning_asset_ack_persistence_failed",
|
|
"receipt_persisted": False,
|
|
"learning_receipt_refs": {},
|
|
"runtime_closure_verified": False,
|
|
}
|
|
return {
|
|
**envelope,
|
|
"status": "learning_asset_acks_recorded",
|
|
"receipt_persisted": persisted,
|
|
"learning_receipt_refs": merged,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
|
|
async def read_latest_for_incident(
|
|
self,
|
|
*,
|
|
project_id: str,
|
|
incident_id: str,
|
|
) -> dict[str, Any] | None:
|
|
if not incident_id:
|
|
return None
|
|
try:
|
|
async with get_db_context(project_id or "awoooi") as db:
|
|
result = await db.execute(
|
|
select(
|
|
AwoooPRunState.run_id,
|
|
AwoooPRunState.state,
|
|
AwoooPRunState.trace_id,
|
|
AwoooPRunState.error_detail,
|
|
)
|
|
.where(
|
|
AwoooPRunState.project_id == (project_id or "awoooi"),
|
|
AwoooPRunState.agent_id == AGENT99_DISPATCH_AGENT_ID,
|
|
AwoooPRunState.trigger_ref.like(f"agent99:{incident_id}:%"),
|
|
)
|
|
.order_by(AwoooPRunState.created_at.desc())
|
|
.limit(1)
|
|
)
|
|
row = result.one_or_none()
|
|
envelope = _parse_envelope(row.error_detail if row else None)
|
|
if not isinstance(envelope, dict) or row is None:
|
|
return None
|
|
return {
|
|
**envelope,
|
|
"run_state": str(row.state),
|
|
"run_id": str(row.run_id),
|
|
"trace_id": str(row.trace_id or ""),
|
|
}
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_dispatch_receipt_read_failed",
|
|
incident_id=incident_id,
|
|
error=str(exc),
|
|
)
|
|
return None
|
|
|
|
async def list_reconcilable(
|
|
self,
|
|
*,
|
|
project_id: str = "awoooi",
|
|
limit: int = 20,
|
|
) -> list[dict[str, Any]]:
|
|
"""List durable deliveries that need outcome or learning reconciliation."""
|
|
|
|
try:
|
|
async with get_db_context(project_id or "awoooi") as db:
|
|
result = await db.execute(
|
|
select(
|
|
AwoooPRunState.run_id,
|
|
AwoooPRunState.state,
|
|
AwoooPRunState.error_detail,
|
|
)
|
|
.where(
|
|
AwoooPRunState.project_id == (project_id or "awoooi"),
|
|
AwoooPRunState.agent_id == AGENT99_DISPATCH_AGENT_ID,
|
|
AwoooPRunState.state.in_([
|
|
"pending",
|
|
"running",
|
|
"waiting_tool",
|
|
]),
|
|
)
|
|
# Prioritize observed delivery/outcome lanes; pending crash
|
|
# recovery candidates are polled after them.
|
|
.order_by(
|
|
(AwoooPRunState.state == "pending").asc(),
|
|
AwoooPRunState.created_at.asc(),
|
|
)
|
|
.limit(max(1, min(int(limit), 100)))
|
|
)
|
|
rows = result.all()
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_dispatch_reconciliation_list_failed",
|
|
project_id=project_id,
|
|
error=str(exc),
|
|
)
|
|
return []
|
|
|
|
items: list[dict[str, Any]] = []
|
|
for row in rows:
|
|
envelope = _parse_envelope(row.error_detail)
|
|
public_identity = (
|
|
envelope.get("identity")
|
|
if isinstance(envelope, dict)
|
|
and isinstance(envelope.get("identity"), dict)
|
|
else None
|
|
)
|
|
if not isinstance(public_identity, dict):
|
|
continue
|
|
try:
|
|
identity = agent99_dispatch_identity_from_public(public_identity)
|
|
except ValueError:
|
|
logger.warning(
|
|
"agent99_dispatch_reconciliation_identity_invalid",
|
|
run_id=str(row.run_id),
|
|
)
|
|
continue
|
|
items.append({
|
|
"identity": identity,
|
|
"run_state": str(row.state),
|
|
"receipt": envelope,
|
|
})
|
|
return items
|
|
|
|
|
|
def _parse_envelope(value: Any) -> dict[str, Any] | None:
|
|
if isinstance(value, dict):
|
|
return value
|
|
if not value:
|
|
return None
|
|
try:
|
|
parsed = json.loads(str(value))
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
return None
|
|
return parsed if isinstance(parsed, dict) else None
|
|
|
|
|
|
_ledger = PostgresAgent99DispatchLedger()
|
|
|
|
|
|
def get_agent99_dispatch_ledger() -> PostgresAgent99DispatchLedger:
|
|
return _ledger
|
|
|
|
|
|
async def read_agent99_dispatch_receipt(
|
|
*,
|
|
project_id: str,
|
|
incident_id: str,
|
|
) -> dict[str, Any] | None:
|
|
return await _ledger.read_latest_for_incident(
|
|
project_id=project_id,
|
|
incident_id=incident_id,
|
|
)
|
|
|
|
|
|
async def list_agent99_reconcilable_dispatches(
|
|
*,
|
|
project_id: str = "awoooi",
|
|
limit: int = 20,
|
|
) -> list[dict[str, Any]]:
|
|
return await _ledger.list_reconcilable(
|
|
project_id=project_id,
|
|
limit=limit,
|
|
)
|
|
|
|
|
|
async def record_agent99_verifier_receipt(
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
outcome_receipt: dict[str, Any],
|
|
evidence_refs: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
return await _ledger.record_verifier(
|
|
identity=identity,
|
|
outcome_receipt=outcome_receipt,
|
|
evidence_refs=evidence_refs,
|
|
)
|
|
|
|
|
|
async def record_agent99_learning_writeback(
|
|
*,
|
|
identity: Agent99DispatchIdentity,
|
|
receipt_refs: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
result = await _ledger.record_learning_writeback(
|
|
identity=identity,
|
|
receipt_refs=receipt_refs,
|
|
)
|
|
if result.get("postgres_terminal_bundle_committed") is not True:
|
|
return result
|
|
|
|
projection_acknowledged = False
|
|
try:
|
|
# PostgreSQL is source-of-truth. Working memory is refreshed only
|
|
# after the terminal transaction above has committed successfully.
|
|
from src.services.incident_service import get_incident_service
|
|
|
|
incident_service = get_incident_service()
|
|
incident = await incident_service.get_from_episodic_memory(
|
|
identity.incident_id,
|
|
project_id=identity.project_id,
|
|
)
|
|
projection_acknowledged = bool(
|
|
incident is not None
|
|
and await incident_service.save_to_working_memory(incident)
|
|
)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"agent99_incident_redis_projection_pending",
|
|
incident_id=identity.incident_id,
|
|
run_id=str(identity.run_id),
|
|
error=str(exc),
|
|
)
|
|
return {
|
|
**result,
|
|
"redis_projection_pending": not projection_acknowledged,
|
|
"redis_projection_acknowledged": projection_acknowledged,
|
|
}
|