"""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 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, 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() @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_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, }