Scope boundary

Owns immutable operation or business-intent identity, atomic action ledger, durable outbox, trusted provider binding, duplicate suppression, at-most-once effect handling, and explicit uncertain-outcome reconciliation or compensation paths for state-changing tools.

AID-H-018.006: AID-H-018.008 prevents repeated or inconsistent external effects; AID-H-018.006 proves the exact action is freshly authorized immediately before dispatch. Authorization evidence may refresh without changing the stable operation identity.

AID-R-007: AID-H-018.008 designs dispatch-time idempotency and records effect state; AID-R-007 performs post-incident authoritative readback, cancellation, reversal, compensation, and final disposition. Recovery consumes the ledger but does not redesign the original dispatch guarantee.

AID-H-004.003: AID-H-018.008 prevents duplicate external side effects; AID-H-004.003 protects peer identity, payload integrity, task state, and replay resistance in agent communication. One message can be accepted once yet still cause duplicate provider effects after retry or worker failure.

AID-H-004.005: AID-H-018.008 suppresses duplicate business effects and reconciles uncertain provider outcomes; AID-H-004.005 rejects duplicate, stale, altered, or context-misbound messages before receiver processing. An accepted message can still produce an ambiguous external effect, so replay protection does not replace the action ledger.

Implementation guidance

• AID-H-018.008-G001: Derive an action key from trusted execution context, reserve it atomically with a durable outbox record, and return the recorded result for duplicate deliveries instead of dispatching again.
• AID-H-018.008-G002: Dispatch a reserved action once, propagate the same key to providers that support idempotency, and reconcile unknown outcomes before retrying or compensating.

AID-H-018.008-G001

Concept:

The model, caller, approval UI, and provider must not choose the effect identity. A trusted workflow service creates one immutable operation ID for one business intent. That ID and its provider idempotency key survive re-approval and policy rotation; approval and policy versions remain evidence only. The ledger binds canonical arguments, tenant, tool, and a destination resolved from server-owned configuration. A second request for the same workflow step either returns the existing operation or fails if any immutable effect field changed.

Step 1: Create the durable ledger, authorization evidence, and outbox

-- File: db/agent_action_ledger.sql
CREATE TABLE agent_action_ledger (
  operation_id uuid PRIMARY KEY,
  effect_key text NOT NULL UNIQUE,
  principal_id text NOT NULL,
  task_id text NOT NULL,
  workflow_step_id text NOT NULL,
  tool_name text NOT NULL,
  tenant_id text NOT NULL,
  provider_id text NOT NULL,
  provider_endpoint text NOT NULL,
  canonical_args_sha256 text NOT NULL,
  arguments_json jsonb NOT NULL,
  current_approval_id text NOT NULL,
  current_policy_version text NOT NULL,
  status text NOT NULL CHECK (status IN (
    'pending', 'executing', 'succeeded', 'unknown',
    'needs_review', 'compensating', 'compensated', 'failed'
  )),
  provider_receipt text,
  result_json jsonb,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (principal_id, task_id, workflow_step_id)
);

CREATE TABLE agent_action_outbox (
  operation_id uuid PRIMARY KEY REFERENCES agent_action_ledger(operation_id),
  claimed_at timestamptz,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE agent_action_authorization_evidence (
  operation_id uuid NOT NULL REFERENCES agent_action_ledger(operation_id),
  approval_id text NOT NULL,
  policy_version text NOT NULL,
  decision_sha256 text NOT NULL,
  authorized_at timestamptz NOT NULL,
  expires_at timestamptz NOT NULL,
  PRIMARY KEY (operation_id, approval_id, policy_version, decision_sha256)
);

Step 2: Mint one operation and bind its destination atomically

# File: tool_runtime/reserve_action.py
from __future__ import annotations

import hashlib
import hmac
import json
import os
import uuid

import psycopg
from psycopg.types.json import Jsonb


EFFECT_KEY_SECRET = os.environ["EFFECT_KEY_SECRET"].encode("utf-8")

# This registry is deployed with the trusted dispatcher, not supplied by the model.
TOOL_PROVIDER_REGISTRY = {
    "issue_refund": {
        "provider_id": "billing-prod-v3",
        "endpoint": "https://billing.internal.example.org/v3/refunds",
        "credential_env": "BILLING_TOOL_TOKEN",
        "allowed_tenants": {"tenant-a", "tenant-b"},
        "idempotency_contract": {
            "mechanism": "header", "name": "Idempotency-Key",
        },
        "reconciliation_key": "effect_key",
    },
}


def canonical_json(value: dict) -> bytes:
    if not isinstance(value, dict):
        raise ValueError("tool arguments must be one JSON object")
    return json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    ).encode("utf-8")


def resolve_provider(tool_name: str, tenant_id: str) -> dict:
    provider = TOOL_PROVIDER_REGISTRY.get(tool_name)
    if provider is None or tenant_id not in provider["allowed_tenants"]:
        raise PermissionError("tool/provider/tenant route is not approved")
    return provider


def derive_effect_key(operation_id: str) -> str:
    material = ("aidefend-effect-v1:" + operation_id).encode("ascii")
    return hmac.new(EFFECT_KEY_SECRET, material, hashlib.sha256).hexdigest()


def reserve_operation(
    conn: psycopg.Connection,
    trusted_context: dict,
    tool_name: str,
    arguments: dict,
) -> dict:
    # trusted_context comes from authenticated workflow state, never request/model text.
    principal_id = trusted_context["principal_id"]
    task_id = trusted_context["task_id"]
    workflow_step_id = trusted_context["workflow_step_id"]
    tenant_id = trusted_context["tenant_id"]
    approval_id = trusted_context["approval_id"]
    policy_version = trusted_context["policy_version"]
    provider = resolve_provider(tool_name, tenant_id)
    args_bytes = canonical_json(arguments)
    args_sha256 = hashlib.sha256(args_bytes).hexdigest()
    proposed_operation_id = str(uuid.uuid4())
    proposed_effect_key = derive_effect_key(proposed_operation_id)

    with conn.transaction():
        inserted = conn.execute(
            """
            INSERT INTO agent_action_ledger (
              operation_id, effect_key, principal_id, task_id, workflow_step_id,
              tool_name, tenant_id, provider_id, provider_endpoint,
              canonical_args_sha256, arguments_json, current_approval_id,
              current_policy_version, status
            ) VALUES (
              %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'pending'
            )
            ON CONFLICT (principal_id, task_id, workflow_step_id) DO NOTHING
            RETURNING operation_id
            """,
            (
                proposed_operation_id, proposed_effect_key, principal_id, task_id,
                workflow_step_id, tool_name, tenant_id, provider["provider_id"],
                provider["endpoint"], args_sha256, Jsonb(arguments), approval_id,
                policy_version,
            ),
        ).fetchone()

        row = conn.execute(
            """
            SELECT operation_id::text, effect_key, status, tool_name, tenant_id,
                   canonical_args_sha256, provider_id, provider_endpoint,
                   current_approval_id, current_policy_version,
                   provider_receipt, result_json
            FROM agent_action_ledger
            WHERE principal_id = %s AND task_id = %s AND workflow_step_id = %s
            FOR UPDATE
            """,
            (principal_id, task_id, workflow_step_id),
        ).fetchone()
        if row is None:
            raise RuntimeError("operation reservation disappeared")
        if (
            row[3] != tool_name
            or row[4] != tenant_id
            or row[5] != args_sha256
            or row[6] != provider["provider_id"]
            or row[7] != provider["endpoint"]
        ):
            raise RuntimeError("same operation was replayed with changed effect fields")

        created = inserted is not None
        if created:
            conn.execute(
                "INSERT INTO agent_action_outbox (operation_id) VALUES (%s)",
                (row[0],),
            )
        elif row[2] == "pending" and (
            row[8] != approval_id or row[9] != policy_version
        ):
            conn.execute(
                """
                UPDATE agent_action_ledger
                SET current_approval_id = %s, current_policy_version = %s,
                    updated_at = now()
                WHERE operation_id = %s
                """,
                (approval_id, policy_version, row[0]),
            )

        return {
            "operation_id": row[0],
            "effect_key": row[1],
            "created": created,
            "status": row[2],
            "provider_receipt": row[10],
            "result": row[11],
        }

Action: Let the authenticated workflow service call reserve_operation once per immutable workflow step. Re-delivery with a refreshed approval returns the same operation and effect key; changed arguments, tenant, tool, or destination fail closed. Only the outbox worker may claim the resulting row, and it must re-authorize immediately before external dispatch as shown next.

Keep the action safely bounded

Reconcile the complete population of every logical workflow step and every duplicate delivery or retry for that step; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Verify safely

A separately credentialed verifier independently replays ledger/outbox transactions and canonicalization with duplicate, concurrent, changed-argument/provider/tenant, crash-before/after-commit, and authorization-refresh fixtures, reads back effective state from the enforcement system, exercises positive, negative, boundary, stale-evidence, and dependency-failure fixtures, and reconciles every eligible item.

AID-H-018.008-G002

Concept:

A database transaction cannot atomically commit an unrelated SaaS or payment effect. Claim the immutable operation, resolve only its ledger-bound provider route, and obtain a fresh authorization decision immediately before dispatch. Never blindly reissue an executing or unknown operation after a timeout or worker crash. If the provider supports idempotency, always reuse the same effect key. Otherwise query the provider by receipt or immutable business reference and route ambiguity to reconciliation.

Worker state transitions and fresh dispatch authorization

# File: tool_runtime/execute_action.py
from __future__ import annotations

import hashlib
import json
import math
import os
import ssl
import subprocess
import tempfile
from pathlib import Path
from datetime import datetime, timezone
from urllib.parse import urlsplit, urlunsplit

import httpx
import psycopg
from psycopg.types.json import Jsonb

from tool_runtime.reserve_action import TOOL_PROVIDER_REGISTRY, resolve_provider


AUTHZ_CA_BUNDLE = os.environ["DISPATCH_AUTHZ_CA_BUNDLE"]
AUTHZ_CLIENT_CERT = os.environ["DISPATCH_AUTHZ_CLIENT_CERT"]
AUTHZ_CLIENT_KEY = os.environ["DISPATCH_AUTHZ_CLIENT_KEY"]
PROFILE_FIELDS = {
    "profile_id", "authorization_url", "maximum_authorization_age_seconds",
    "maximum_authorization_ttl_seconds", "maximum_clock_skew_seconds",
    "authorization_http_timeout_seconds", "provider_http_timeout_seconds",
    "authorization_response_max_bytes", "provider_response_max_bytes",
}


def strict_json(raw: bytes, label: str) -> dict:
    def reject_duplicates(pairs: list[tuple[str, object]]) -> dict:
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"{label} contains duplicate key: {key}")
            value[key] = item
        return value

    def reject_constant(value: str) -> None:
        raise ValueError(f"{label} contains non-finite constant: {value}")

    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=reject_duplicates,
        parse_constant=reject_constant,
    )
    if not isinstance(value, dict):
        raise ValueError(f"{label} root must be an object")
    return value


def canonical_https(value: str, label: str) -> str:
    parts = urlsplit(value)
    if (
        parts.scheme != "https" or not parts.hostname
        or parts.username is not None or parts.password is not None
        or parts.port is not None or parts.query or parts.fragment
        or "\\" in value or parts.netloc != parts.hostname
    ):
        raise ValueError(f"{label} must be an exact HTTPS URL")
    canonical = urlunsplit(("https", parts.hostname, parts.path, "", ""))
    if value != canonical:
        raise ValueError(f"{label} is not canonical")
    return value


def load_verified_profile() -> dict:
    profile_path = Path(os.environ["VERIFIED_DISPATCH_PROFILE_PATH"])
    profile_bytes = profile_path.read_bytes()
    verify_timeout = float(os.environ["DISPATCH_PROFILE_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("profile verification timeout is invalid")
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(prefix="dispatch-profile-", suffix=".json", delete=False) as handle:
            handle.write(profile_bytes)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", os.environ["DISPATCH_PROFILE_PUBLIC_KEY"],
                "--bundle", os.environ["DISPATCH_PROFILE_SIGNATURE"], str(snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=verify_timeout,
        )
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)
    profile = strict_json(profile_bytes, "dispatch profile")
    if set(profile) != PROFILE_FIELDS:
        raise RuntimeError("dispatch profile schema mismatch")
    return profile


def required_integer(value: object, name: str, *, allow_zero: bool = False) -> int:
    minimum = 0 if allow_zero else 1
    if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
        raise RuntimeError(name + " must be a strict integer within policy bounds")
    return value


def required_positive_number(value: object, name: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise RuntimeError(name + " must be numeric")
    result = float(value)
    if not math.isfinite(result) or result <= 0:
        raise RuntimeError(name + " must be finite and positive")
    return result


DISPATCH_PROFILE = load_verified_profile()
if not isinstance(DISPATCH_PROFILE["profile_id"], str) or not DISPATCH_PROFILE["profile_id"]:
    raise RuntimeError("dispatch profile ID is invalid")
AUTHZ_URL = canonical_https(DISPATCH_PROFILE["authorization_url"], "authorization URL")
MAX_AUTHORIZATION_AGE_SECONDS = required_integer(
    DISPATCH_PROFILE["maximum_authorization_age_seconds"], "maximum authorization age"
)
MAX_AUTHORIZATION_TTL_SECONDS = required_integer(
    DISPATCH_PROFILE["maximum_authorization_ttl_seconds"], "maximum authorization TTL"
)
MAX_CLOCK_SKEW_SECONDS = required_integer(
    DISPATCH_PROFILE["maximum_clock_skew_seconds"], "maximum clock skew", allow_zero=True
)
AUTHZ_HTTP_TIMEOUT_SECONDS = required_positive_number(
    DISPATCH_PROFILE["authorization_http_timeout_seconds"], "authorization timeout"
)
PROVIDER_HTTP_TIMEOUT_SECONDS = required_positive_number(
    DISPATCH_PROFILE["provider_http_timeout_seconds"], "provider timeout"
)
AUTHZ_RESPONSE_MAX_BYTES = required_integer(
    DISPATCH_PROFILE["authorization_response_max_bytes"], "authorization response cap"
)
PROVIDER_RESPONSE_MAX_BYTES = required_integer(
    DISPATCH_PROFILE["provider_response_max_bytes"], "provider response cap"
)
if (
    min(
        MAX_AUTHORIZATION_AGE_SECONDS, MAX_AUTHORIZATION_TTL_SECONDS,
        AUTHZ_RESPONSE_MAX_BYTES, PROVIDER_RESPONSE_MAX_BYTES,
    ) <= 0
    or MAX_CLOCK_SKEW_SECONDS < 0
    or any(
        not math.isfinite(value) or value <= 0
        for value in (AUTHZ_HTTP_TIMEOUT_SECONDS, PROVIDER_HTTP_TIMEOUT_SECONDS)
    )
):
    raise RuntimeError("signed dispatch-profile bounds are invalid")


def parse_time(value: object) -> datetime:
    if not isinstance(value, str) or not value:
        raise ValueError("authorization timestamp must be a non-empty string")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("authorization timestamp must include a timezone")
    return parsed.astimezone(timezone.utc)


def canonical_json(value: dict) -> bytes:
    return json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    ).encode("utf-8")


def bounded_json(response: httpx.Response, maximum_bytes: int, label: str) -> tuple[dict, bytes]:
    length = response.headers.get("content-length")
    if length is not None and (not length.isdigit() or int(length) > maximum_bytes):
        raise ValueError(f"{label} exceeds the signed byte cap")
    raw = bytearray()
    for chunk in response.iter_bytes():
        raw.extend(chunk)
        if len(raw) > maximum_bytes:
            raise ValueError(f"{label} exceeds the signed byte cap")
    raw_bytes = bytes(raw)
    return strict_json(raw_bytes, label), raw_bytes


def claim_pending(conn: psycopg.Connection) -> dict | None:
    with conn.transaction():
        row = conn.execute(
            """
            SELECT a.operation_id::text, a.effect_key, a.principal_id, a.task_id,
                   a.workflow_step_id, a.tool_name, a.tenant_id, a.provider_id,
                   a.provider_endpoint, a.canonical_args_sha256, a.arguments_json,
                   a.current_approval_id, a.current_policy_version
            FROM agent_action_outbox o
            JOIN agent_action_ledger a USING (operation_id)
            WHERE a.status = 'pending' AND o.claimed_at IS NULL
            ORDER BY o.created_at
            FOR UPDATE OF o, a SKIP LOCKED
            LIMIT 1
            """
        ).fetchone()
        if row is None:
            return None
        conn.execute(
            "UPDATE agent_action_outbox SET claimed_at = now() WHERE operation_id = %s",
            (row[0],),
        )
        conn.execute(
            """
            UPDATE agent_action_ledger
            SET status = 'executing', updated_at = now()
            WHERE operation_id = %s AND status = 'pending'
            """,
            (row[0],),
        )
        return {
            "operation_id": row[0],
            "effect_key": row[1],
            "principal_id": row[2],
            "task_id": row[3],
            "workflow_step_id": row[4],
            "tool_name": row[5],
            "tenant_id": row[6],
            "provider_id": row[7],
            "provider_endpoint": row[8],
            "canonical_args_sha256": row[9],
            "arguments": row[10],
            "approval_id": row[11],
            "policy_version": row[12],
        }


def require_bound_provider(action: dict) -> dict:
    provider = resolve_provider(action["tool_name"], action["tenant_id"])
    endpoint = canonical_https(provider["endpoint"], "provider endpoint")
    idempotency = provider.get("idempotency_contract")
    if (
        provider["provider_id"] != action["provider_id"]
        or endpoint != action["provider_endpoint"]
        or idempotency != {"mechanism": "header", "name": "Idempotency-Key"}
        or provider.get("reconciliation_key") != "effect_key"
    ):
        raise PermissionError("provider route lacks the reserved idempotency contract")
    return provider


def build_authz_client() -> httpx.Client:
    context = ssl.create_default_context(cafile=AUTHZ_CA_BUNDLE)
    context.load_cert_chain(AUTHZ_CLIENT_CERT, AUTHZ_CLIENT_KEY)
    return httpx.Client(
        verify=context, timeout=AUTHZ_HTTP_TIMEOUT_SECONDS,
        follow_redirects=False, trust_env=False,
    )


def require_fresh_authorization(action: dict) -> tuple[dict, str]:
    request_body = {
        "operation_id": action["operation_id"],
        "effect_key": action["effect_key"],
        "principal_id": action["principal_id"],
        "task_id": action["task_id"],
        "workflow_step_id": action["workflow_step_id"],
        "tool_name": action["tool_name"],
        "tenant_id": action["tenant_id"],
        "provider_id": action["provider_id"],
        "canonical_args_sha256": action["canonical_args_sha256"],
        "approval_id": action["approval_id"],
        "policy_version": action["policy_version"],
    }
    with build_authz_client() as client:
        with client.stream("POST", AUTHZ_URL, json=request_body) as response:
            response.raise_for_status()
            decision, _ = bounded_json(response, AUTHZ_RESPONSE_MAX_BYTES, "authorization response")
    expected_fields = set(request_body) | {"decision", "authorized_at", "expires_at"}
    if set(decision) != expected_fields or decision.get("decision") != "allow":
        raise PermissionError("fresh dispatch authorization denied")

    for field, value in request_body.items():
        if decision.get(field) != value:
            raise PermissionError("authorization binding mismatch: " + field)
    now = datetime.now(timezone.utc)
    authorized_at = parse_time(decision["authorized_at"])
    expires_at = parse_time(decision["expires_at"])
    age = (now - authorized_at).total_seconds()
    ttl = (expires_at - authorized_at).total_seconds()
    if age < -MAX_CLOCK_SKEW_SECONDS or age > MAX_AUTHORIZATION_AGE_SECONDS:
        raise PermissionError("authorization decision is not fresh")
    if expires_at <= now or ttl <= 0 or ttl > MAX_AUTHORIZATION_TTL_SECONDS:
        raise PermissionError("authorization decision lifetime is invalid")
    decision_sha256 = hashlib.sha256(canonical_json(decision)).hexdigest()
    return decision, decision_sha256


def record_fresh_authorization(
    conn: psycopg.Connection,
    action: dict,
    decision: dict,
    decision_sha256: str,
) -> None:
    with conn.transaction():
        row = conn.execute(
            """
            SELECT status, current_approval_id, current_policy_version
            FROM agent_action_ledger WHERE operation_id = %s FOR UPDATE
            """,
            (action["operation_id"],),
        ).fetchone()
        if row != ("executing", action["approval_id"], action["policy_version"]):
            raise PermissionError("operation authorization context changed before dispatch")
        conn.execute(
            """
            INSERT INTO agent_action_authorization_evidence (
              operation_id, approval_id, policy_version, decision_sha256,
              authorized_at, expires_at
            ) VALUES (%s, %s, %s, %s, %s, %s)
            ON CONFLICT DO NOTHING
            """,
            (
                action["operation_id"], action["approval_id"],
                action["policy_version"], decision_sha256,
                decision["authorized_at"], decision["expires_at"],
            ),
        )


def dispatch_under_authorization_lock(
    conn: psycopg.Connection, action: dict, decision_sha256: str,
    credential: str, provider: dict,
) -> tuple[dict, str | None, str]:
    with conn.transaction():
        row = conn.execute(
            """
            SELECT a.status, a.current_approval_id, a.current_policy_version,
                   e.decision_sha256, e.authorized_at, e.expires_at
            FROM agent_action_ledger a
            JOIN agent_action_authorization_evidence e
              ON e.operation_id = a.operation_id AND e.decision_sha256 = %s
            WHERE a.operation_id = %s
            FOR UPDATE OF a
            """,
            (decision_sha256, action["operation_id"]),
        ).fetchone()
        if row is None or row[:4] != (
            "executing", action["approval_id"], action["policy_version"], decision_sha256,
        ):
            raise PermissionError("dispatch authorization context changed")
        now = datetime.now(timezone.utc)
        authorized_at = row[4].astimezone(timezone.utc)
        expires_at = row[5].astimezone(timezone.utc)
        age = (now - authorized_at).total_seconds()
        if (
            age < -MAX_CLOCK_SKEW_SECONDS
            or age > MAX_AUTHORIZATION_AGE_SECONDS
            or expires_at <= now
        ):
            raise PermissionError("recorded authorization is no longer fresh")

        with httpx.Client(
            timeout=PROVIDER_HTTP_TIMEOUT_SECONDS, follow_redirects=False, trust_env=False
        ) as client:
            with client.stream(
                "POST", action["provider_endpoint"], json=action["arguments"],
                headers={
                    "Authorization": "Bearer " + credential,
                    provider["idempotency_contract"]["name"]: action["effect_key"],
                },
            ) as response:
                response.raise_for_status()
                request_id = response.headers.get("x-request-id")
                body, raw_body = bounded_json(
                    response, PROVIDER_RESPONSE_MAX_BYTES, "provider response"
                )
        return body, request_id, hashlib.sha256(raw_body).hexdigest()


def transition(
    conn: psycopg.Connection,
    operation_id: str,
    status: str,
    result: dict,
    provider_receipt: str | None = None,
) -> None:
    with conn.transaction():
        updated = conn.execute(
            """
            UPDATE agent_action_ledger
            SET status = %s, result_json = %s, provider_receipt = %s,
                updated_at = now()
            WHERE operation_id = %s AND status = 'executing'
            """,
            (status, Jsonb(result), provider_receipt, operation_id),
        )
        if updated.rowcount != 1:
            raise RuntimeError("operation state changed concurrently")


def execute_reserved(conn: psycopg.Connection, action: dict) -> None:
    try:
        provider = require_bound_provider(action)
        decision, decision_sha256 = require_fresh_authorization(action)
        record_fresh_authorization(conn, action, decision, decision_sha256)
        credential = os.environ[provider["credential_env"]]
    except Exception as exc:
        transition(
            conn,
            action["operation_id"],
            "needs_review",
            {"stage": "pre_dispatch_authorization", "error_type": type(exc).__name__},
        )
        return

    try:
        body, request_id, body_sha256 = dispatch_under_authorization_lock(
            conn, action, decision_sha256, credential, provider
        )
    except (httpx.HTTPError, ValueError, PermissionError) as exc:
        transition(
            conn,
            action["operation_id"],
            "unknown",
            {"stage": "provider_dispatch", "error_type": type(exc).__name__},
        )
        return

    receipt = body.get("operation_id") or request_id
    if not isinstance(receipt, str) or not receipt:
        transition(
            conn,
            action["operation_id"],
            "needs_review",
            {"stage": "provider_response", "body_sha256": body_sha256, "issue": "receipt_missing"},
        )
        return
    transition(
        conn,
        action["operation_id"],
        "succeeded",
        {"stage": "provider_response", "body_sha256": body_sha256},
        provider_receipt=receipt,
    )

Reconciliation and compensation rules

• For unknown, query the ledger-bound provider using the same effect key, provider receipt, or immutable business reference. Mark succeeded only from provider evidence.
• If the provider authoritatively proves no effect occurred, require reviewed authorization before creating a new outbox attempt for the same operation and effect key.
• If a partial multi-step effect occurred, create a separate server-minted compensation operation that references the original operation ID, has its own stable effect key, and passes fresh authorization. Do not silently invent an inverse action.
• Never reclaim stale executing rows by age alone; first reconcile them because a worker may have crashed after the provider committed.

Verify safely

An independent job compares provider operations to stable effect keys and verifies that one operation never changed arguments, tenant, tool, or provider destination. It also replays the signed dispatch-profile verification and checks the stored authorization-decision and bounded provider-body digests without retaining an authenticated provider body that may contain secrets.

Keep the action safely bounded

Reconcile the complete population of every outbox claim, provider attempt, response/timeout, reconciliation check, retry, and compensation for the operation; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Open-source tools

• PostgreSQL
• Valkey
• Redis 8+ (AGPL-3.0)
• Temporal
• Apache Kafka
• Debezium
• OpenTelemetry

Commercial / hosted tools

• Temporal Cloud
• AWS Step Functions
• Azure Durable Functions
• Google Cloud Workflows
• Confluent Cloud

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• Integration Risks (L7) (transaction-safe integration prevents duplicate or ambiguous downstream effects)
• Repudiation (L7) (durable action receipts make agent side effects traceable)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI08:2026 Cascading Failures (transactional execution and compensation stop duplicate effects from propagating)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-4.3.4 Replay Exploitation

Google Secure AI Framework 2.0 - Risks

• IIC: Insecure Integrated Component (transaction-safe dispatch hardens agent-to-service integration)

Databricks AI Security Framework 3.0

• Agents - Core 13.8: Repudiation & Untraceability