Connecting LLMs to APIs and Business Workflows Safely

{
  "fixture": "SYNTHETIC EDUCATIONAL FIXTURE — LLM action gateway; no model, network, or real API",
  "version": 1,
  "actors": {
    "ops-a": {
      "tenant_id": "tenant-a",
      "permissions": [
        "customer:read",
        "customer:write",
        "customer:note",
        "case:close"
      ]
    },
    "ops-b": {
      "tenant_id": "tenant-b",
      "permissions": [
        "customer:read",
        "customer:write"
      ]
    }
  },
  "state": {
    "customers": {
      "cust-104": {
        "tenant_id": "tenant-a",
        "priority": "normal",
        "version": 3
      },
      "cust-900": {
        "tenant_id": "tenant-a",
        "priority": "normal",
        "version": 7
      },
      "cust-205": {
        "tenant_id": "tenant-b",
        "priority": "low",
        "version": 2
      }
    },
    "support_cases": {
      "case-22": {
        "tenant_id": "tenant-a",
        "status": "open",
        "legal_hold": false,
        "version": 5
      },
      "case-99": {
        "tenant_id": "tenant-a",
        "status": "open",
        "legal_hold": true,
        "version": 2
      }
    }
  },
  "cases": [
    {
      "id": "valid-read",
      "actor_id": "ops-a",
      "proposal": {
        "action": "read_customer_summary",
        "arguments": {
          "customer_id": "cust-104"
        },
        "idempotency_key": null,
        "expected_version": 3
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "read_only_completed"
    },
    {
      "id": "valid-reversible-write",
      "actor_id": "ops-a",
      "proposal": {
        "action": "update_customer_priority",
        "arguments": {
          "customer_id": "cust-104",
          "priority": "high"
        },
        "idempotency_key": "priority:cust-104:v3",
        "expected_version": 3
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "write_completed"
    },
    {
      "id": "schema-invalid-extra-field",
      "actor_id": "ops-a",
      "proposal": {
        "action": "update_customer_priority",
        "arguments": {
          "customer_id": "cust-104",
          "priority": "high"
        },
        "idempotency_key": "schema:cust-104:v3",
        "expected_version": 3,
        "unexpected_field": true
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "schema_rejected"
    },
    {
      "id": "unknown-action",
      "actor_id": "ops-a",
      "proposal": {
        "action": "set_customer_score",
        "arguments": {
          "customer_id": "cust-104",
          "score": 5
        },
        "idempotency_key": "score:cust-104:v4",
        "expected_version": 4
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "unknown_action"
    },
    {
      "id": "tenant-authorization-denial",
      "actor_id": "ops-b",
      "proposal": {
        "action": "update_customer_priority",
        "arguments": {
          "customer_id": "cust-104",
          "priority": "low"
        },
        "idempotency_key": "denied:cust-104:v4",
        "expected_version": 4
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "authorization_denied"
    },
    {
      "id": "missing-customer-semantic-failure",
      "actor_id": "ops-a",
      "proposal": {
        "action": "read_customer_summary",
        "arguments": {
          "customer_id": "cust-404"
        },
        "idempotency_key": null,
        "expected_version": 1
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "semantic_rejected"
    },
    {
      "id": "legal-hold-policy-block",
      "actor_id": "ops-a",
      "proposal": {
        "action": "close_support_case",
        "arguments": {
          "case_id": "case-99",
          "reason": "Synthetic duplicate case"
        },
        "idempotency_key": "close:case-99:v2",
        "expected_version": 2
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "policy_blocked"
    },
    {
      "id": "sensitive-note-needs-approval",
      "actor_id": "ops-a",
      "proposal": {
        "action": "add_internal_note",
        "arguments": {
          "customer_id": "cust-104",
          "note": "Synthetic follow-up note."
        },
        "idempotency_key": "note:cust-104:v4",
        "expected_version": 4
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "approval_required"
    },
    {
      "id": "same-operation-replay",
      "actor_id": "ops-a",
      "proposal": {
        "action": "update_customer_priority",
        "arguments": {
          "customer_id": "cust-104",
          "priority": "high"
        },
        "idempotency_key": "priority:cust-104:v3",
        "expected_version": 3
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "idempotent_replay"
    },
    {
      "id": "same-key-changed-payload",
      "actor_id": "ops-a",
      "proposal": {
        "action": "update_customer_priority",
        "arguments": {
          "customer_id": "cust-104",
          "priority": "low"
        },
        "idempotency_key": "priority:cust-104:v3",
        "expected_version": 3
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "idempotency_conflict"
    },
    {
      "id": "retryable-read-dependency-failure",
      "actor_id": "ops-a",
      "proposal": {
        "action": "read_customer_summary",
        "arguments": {
          "customer_id": "cust-900"
        },
        "idempotency_key": null,
        "expected_version": 7
      },
      "approval_reference": null,
      "dependency_results": [
        "temporary_unavailable",
        "temporary_unavailable"
      ],
      "expected": "dependency_failure_retryable"
    },
    {
      "id": "nonretryable-read-dependency-failure",
      "actor_id": "ops-a",
      "proposal": {
        "action": "read_customer_summary",
        "arguments": {
          "customer_id": "cust-900"
        },
        "idempotency_key": null,
        "expected_version": 7
      },
      "approval_reference": null,
      "dependency_results": [
        "invalid_request"
      ],
      "expected": "dependency_failure_nonretryable"
    },
    {
      "id": "stale-version",
      "actor_id": "ops-a",
      "proposal": {
        "action": "update_customer_priority",
        "arguments": {
          "customer_id": "cust-104",
          "priority": "normal"
        },
        "idempotency_key": "stale:cust-104:v3",
        "expected_version": 3
      },
      "approval_reference": null,
      "dependency_results": [
        "success"
      ],
      "expected": "stale_state"
    },
    {
      "id": "invalid-read-response",
      "actor_id": "ops-a",
      "proposal": {
        "action": "read_customer_summary",
        "arguments": {
          "customer_id": "cust-900"
        },
        "idempotency_key": null,
        "expected_version": 7
      },
      "approval_reference": null,
      "dependency_results": [
        "invalid_response"
      ],
      "expected": "response_validation_failed"
    }
  ]
}

from __future__ import annotations

import copy
import hashlib
import json
import sys
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any


LABEL = "SYNTHETIC EDUCATIONAL FIXTURE — LLM action gateway; no model, network, or real API"
FIXTURE_PATH = Path(__file__).with_name("llm_action_gateway_fixture.json")
FIXTURE_FIELDS = {"fixture", "version", "actors", "state", "cases"}
CASE_FIELDS = {"id", "actor_id", "proposal", "approval_reference", "dependency_results", "expected"}
PROPOSAL_FIELDS = {"action", "arguments", "idempotency_key", "expected_version"}
DEPENDENCY_RESULTS = {"success", "temporary_unavailable", "invalid_request", "invalid_response"}
REQUIRED_OUTCOMES = {
    "read_only_completed",
    "write_completed",
    "schema_rejected",
    "unknown_action",
    "authorization_denied",
    "semantic_rejected",
    "policy_blocked",
    "approval_required",
    "idempotent_replay",
    "idempotency_conflict",
    "dependency_failure_retryable",
    "dependency_failure_nonretryable",
    "stale_state",
    "response_validation_failed",
}
MAX_FIXTURE_ATTEMPTS = 2


@dataclass(frozen=True, slots=True)
class ActionSpec:
    name: str
    argument_fields: tuple[str, ...]
    permission: str
    side_effect_class: str
    approval_required: bool
    idempotency_required: bool
    resource_collection: str
    resource_argument: str
    response_fields: frozenset[str]


REGISTRY = {
    "read_customer_summary": ActionSpec(
        "read_customer_summary", ("customer_id",), "customer:read", "READ_ONLY", False, False,
        "customers", "customer_id", frozenset({"customer_id", "priority", "version"}),
    ),
    "update_customer_priority": ActionSpec(
        "update_customer_priority", ("customer_id", "priority"), "customer:write", "REVERSIBLE_WRITE", False, True,
        "customers", "customer_id", frozenset({"customer_id", "priority", "version", "updated"}),
    ),
    "add_internal_note": ActionSpec(
        "add_internal_note", ("customer_id", "note"), "customer:note", "SENSITIVE_WRITE", True, True,
        "customers", "customer_id", frozenset({"customer_id", "note_id", "version", "added"}),
    ),
    "close_support_case": ActionSpec(
        "close_support_case", ("case_id", "reason"), "case:close", "IRREVERSIBLE_WRITE", True, True,
        "support_cases", "case_id", frozenset({"case_id", "status", "version"}),
    ),
}


def load_fixture() -> dict[str, Any]:
    fixture = json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))
    if not isinstance(fixture, dict) or set(fixture) != FIXTURE_FIELDS:
        raise ValueError("fixture fields do not match the exact contract")
    if fixture["fixture"] != LABEL or fixture["version"] != 1:
        raise ValueError("fixture label or version is invalid")
    if not isinstance(fixture["actors"], dict) or not fixture["actors"]:
        raise ValueError("actors must be a non-empty object")
    for actor_id, actor in fixture["actors"].items():
        if not isinstance(actor_id, str) or not isinstance(actor, dict) or set(actor) != {"tenant_id", "permissions"}:
            raise ValueError("actor contract is invalid")
        if not isinstance(actor["tenant_id"], str) or not isinstance(actor["permissions"], list):
            raise ValueError("actor fields have invalid types")
        if not all(isinstance(value, str) and value for value in actor["permissions"]):
            raise ValueError("actor permissions must be non-empty strings")
    if not isinstance(fixture["state"], dict) or set(fixture["state"]) != {"customers", "support_cases"}:
        raise ValueError("state contract is invalid")
    if not isinstance(fixture["cases"], list) or not fixture["cases"]:
        raise ValueError("cases must be a non-empty list")

    seen_ids: set[str] = set()
    expected_outcomes: set[str] = set()
    for case in fixture["cases"]:
        if not isinstance(case, dict) or set(case) != CASE_FIELDS:
            raise ValueError("case fields do not match the exact contract")
        if not isinstance(case["id"], str) or not case["id"] or case["id"] in seen_ids:
            raise ValueError("case IDs must be unique non-empty strings")
        seen_ids.add(case["id"])
        if case["actor_id"] not in fixture["actors"]:
            raise ValueError(f"unknown fixture actor: {case['actor_id']}")
        approval = case["approval_reference"]
        if approval is not None and (not isinstance(approval, str) or not approval):
            raise ValueError("approval_reference must be null or a non-empty string")
        results = case["dependency_results"]
        if not isinstance(results, list) or not 1 <= len(results) <= MAX_FIXTURE_ATTEMPTS:
            raise ValueError("dependency_results must contain one or two events")
        if any(result not in DEPENDENCY_RESULTS for result in results):
            raise ValueError("dependency_results contains an unknown event")
        if case["expected"] not in REQUIRED_OUTCOMES:
            raise ValueError(f"unknown expected outcome: {case['expected']}")
        expected_outcomes.add(case["expected"])
    missing = REQUIRED_OUTCOMES - expected_outcomes
    if missing:
        raise ValueError(f"fixture lacks required outcomes: {sorted(missing)}")
    return fixture


def valid_proposal(proposal: Any) -> bool:
    return (
        isinstance(proposal, dict)
        and set(proposal) == PROPOSAL_FIELDS
        and isinstance(proposal["action"], str)
        and bool(proposal["action"])
        and isinstance(proposal["arguments"], dict)
        and (proposal["idempotency_key"] is None or (isinstance(proposal["idempotency_key"], str) and bool(proposal["idempotency_key"])))
        and type(proposal["expected_version"]) is int
        and proposal["expected_version"] > 0
    )


def valid_arguments(spec: ActionSpec, proposal: dict[str, Any]) -> bool:
    arguments = proposal["arguments"]
    if set(arguments) != set(spec.argument_fields):
        return False
    if not all(isinstance(arguments[field], str) and bool(arguments[field]) for field in spec.argument_fields):
        return False
    if "priority" in arguments and arguments["priority"] not in {"low", "normal", "high"}:
        return False
    if "note" in arguments and len(arguments["note"]) > 120:
        return False
    key = proposal["idempotency_key"]
    return (key is None) if not spec.idempotency_required else isinstance(key, str) and bool(key)


def operation_digest(proposal: dict[str, Any]) -> str:
    operation = {
        "action": proposal["action"],
        "arguments": proposal["arguments"],
        "expected_version": proposal["expected_version"],
    }
    encoded = json.dumps(operation, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


class Gateway:
    def __init__(self, actors: dict[str, Any], state: dict[str, Any]):
        self.actors = actors
        self.state = copy.deepcopy(state)
        self.idempotency: dict[str, dict[str, Any]] = {}
        self.executed_keys: set[str] = set()
        self.approvals: dict[str, dict[str, str]] = {}
        self.audit: list[dict[str, str]] = []
        self.metrics: Counter[str] = Counter()

    def process(self, case: dict[str, Any]) -> str:
        outcome = self._evaluate(case)
        proposal = case["proposal"]
        action = proposal.get("action", "<invalid>") if isinstance(proposal, dict) else "<invalid>"
        self.audit.append({
            "correlation_id": case["id"],
            "actor_id": case["actor_id"],
            "proposed_action": str(action),
            "outcome": outcome,
        })
        return outcome

    def _evaluate(self, case: dict[str, Any]) -> str:
        proposal = case["proposal"]
        if not valid_proposal(proposal):
            return "schema_rejected"
        spec = REGISTRY.get(proposal["action"])
        if spec is None:
            return "unknown_action"
        if not valid_arguments(spec, proposal):
            return "schema_rejected"

        arguments = proposal["arguments"]
        resource = self.state[spec.resource_collection].get(arguments[spec.resource_argument])
        if resource is None:
            return "semantic_rejected"
        actor = self.actors[case["actor_id"]]
        if spec.permission not in actor["permissions"] or resource["tenant_id"] != actor["tenant_id"]:
            return "authorization_denied"
        if spec.name == "close_support_case" and resource["status"] != "open":
            return "semantic_rejected"
        if spec.name == "close_support_case" and resource["legal_hold"] is True:
            return "policy_blocked"

        digest = operation_digest(proposal)
        if spec.approval_required:
            approval = self.approvals.get(case["approval_reference"] or "")
            if approval != {"action": spec.name, "digest": digest, "actor_id": case["actor_id"]}:
                return "approval_required"

        if spec.idempotency_required:
            key = proposal["idempotency_key"]
            assert isinstance(key, str)
            previous = self.idempotency.get(key)
            if previous is not None:
                return "idempotent_replay" if previous["digest"] == digest else "idempotency_conflict"

        if resource["version"] != proposal["expected_version"]:
            return "stale_state"

        response, dependency_outcome = self._call_executor(spec, proposal, resource, case["dependency_results"])
        if dependency_outcome is not None:
            return dependency_outcome
        assert response is not None
        if not self._valid_response(spec, proposal, response):
            return "response_validation_failed"

        if spec.idempotency_required:
            key = proposal["idempotency_key"]
            assert isinstance(key, str)
            self.idempotency[key] = {"digest": digest, "response": response}
            return "write_completed"
        return "read_only_completed"

    def _call_executor(
        self,
        spec: ActionSpec,
        proposal: dict[str, Any],
        resource: dict[str, Any],
        dependency_results: list[str],
    ) -> tuple[dict[str, Any] | None, str | None]:
        for attempt, result in enumerate(dependency_results, start=1):
            if result == "temporary_unavailable":
                retry_is_bounded_and_safe = (
                    spec.side_effect_class == "READ_ONLY"
                    and attempt < len(dependency_results)
                    and attempt < MAX_FIXTURE_ATTEMPTS
                )
                if retry_is_bounded_and_safe:
                    self.metrics["bounded_retries"] += 1
                    continue
                return None, "dependency_failure_retryable"
            if result == "invalid_request":
                return None, "dependency_failure_nonretryable"
            response = self._execute(spec, proposal, resource, authorized=True)
            if result == "invalid_response":
                response.pop("version", None)
            return response, None
        raise AssertionError("dependency sequence exhausted without an outcome")

    def _execute(
        self,
        spec: ActionSpec,
        proposal: dict[str, Any],
        resource: dict[str, Any],
        *,
        authorized: bool,
    ) -> dict[str, Any]:
        if not authorized:
            self.metrics["unauthorized_side_effects"] += 1
            raise AssertionError("executor called without application authorization")
        arguments = proposal["arguments"]
        if spec.name == "read_customer_summary":
            return {"customer_id": arguments["customer_id"], "priority": resource["priority"], "version": resource["version"]}

        key = proposal["idempotency_key"]
        assert isinstance(key, str)
        if key in self.executed_keys:
            self.metrics["duplicate_side_effects"] += 1
        self.executed_keys.add(key)
        self.metrics["executed_side_effects"] += 1

        if spec.name == "update_customer_priority":
            resource["priority"] = arguments["priority"]
            resource["version"] += 1
            return {"customer_id": arguments["customer_id"], "priority": resource["priority"], "version": resource["version"], "updated": True}
        if spec.name == "add_internal_note":
            resource["version"] += 1
            return {"customer_id": arguments["customer_id"], "note_id": f"note-{resource['version']}", "version": resource["version"], "added": True}
        if spec.name == "close_support_case":
            resource["status"] = "closed"
            resource["version"] += 1
            return {"case_id": arguments["case_id"], "status": resource["status"], "version": resource["version"]}
        raise AssertionError("registry executor is missing")

    @staticmethod
    def _valid_response(spec: ActionSpec, proposal: dict[str, Any], response: Any) -> bool:
        if not isinstance(response, dict) or set(response) != set(spec.response_fields):
            return False
        arguments = proposal["arguments"]
        if response.get(spec.resource_argument) != arguments[spec.resource_argument]:
            return False
        if type(response.get("version")) is not int:
            return False
        if spec.name == "read_customer_summary":
            return response["priority"] in {"low", "normal", "high"}
        if spec.name == "update_customer_priority":
            return response["priority"] == arguments["priority"] and response["updated"] is True
        if spec.name == "add_internal_note":
            return isinstance(response["note_id"], str) and response["added"] is True
        return response.get("status") == "closed"


def main() -> int:
    fixture = load_fixture()
    gateway = Gateway(fixture["actors"], fixture["state"])
    outcomes: Counter[str] = Counter()
    failures: list[str] = []

    for case in fixture["cases"]:
        actual = gateway.process(case)
        outcomes[actual] += 1
        if actual != case["expected"]:
            failures.append(f"{case['id']}: expected {case['expected']}, got {actual}")

    passed = len(fixture["cases"]) - len(failures)
    completed = outcomes["read_only_completed"] + outcomes["write_completed"] + outcomes["idempotent_replay"]
    lines = [
        LABEL,
        f"cases={len(fixture['cases'])}",
        f"passed={passed}",
        f"completed={completed}",
        f"read_only_completed={outcomes['read_only_completed']}",
        f"write_completed={outcomes['write_completed']}",
        f"approval_required={outcomes['approval_required']}",
        f"schema_rejected={outcomes['schema_rejected']}",
        f"unknown_action={outcomes['unknown_action']}",
        f"authorization_denied={outcomes['authorization_denied']}",
        f"semantic_rejected={outcomes['semantic_rejected']}",
        f"policy_blocked={outcomes['policy_blocked']}",
        f"idempotent_replay={outcomes['idempotent_replay']}",
        f"idempotency_conflict={outcomes['idempotency_conflict']}",
        f"retryable_dependency_failures={outcomes['dependency_failure_retryable']}",
        f"nonretryable_dependency_failures={outcomes['dependency_failure_nonretryable']}",
        f"stale_state={outcomes['stale_state']}",
        f"response_validation_failed={outcomes['response_validation_failed']}",
        f"bounded_retries={gateway.metrics['bounded_retries']}",
        f"executed_side_effects={gateway.metrics['executed_side_effects']}",
        f"duplicate_side_effects={gateway.metrics['duplicate_side_effects']}",
        f"unauthorized_side_effects={gateway.metrics['unauthorized_side_effects']}",
    ]
    lines.extend(f"FAIL {failure}" for failure in failures)
    if len(gateway.audit) != len(fixture["cases"]):
        lines.append("FAIL audit record count mismatch")
    sys.stdout.buffer.write("\n".join(lines).encode("utf-8"))
    clean = not failures and len(gateway.audit) == len(fixture["cases"])
    safe = gateway.metrics["duplicate_side_effects"] == 0 and gateway.metrics["unauthorized_side_effects"] == 0
    return 0 if clean and safe else 1


if __name__ == "__main__":
    raise SystemExit(main())

SYNTHETIC EDUCATIONAL FIXTURE — LLM action gateway; no model, network, or real API
cases=14
passed=14
completed=3
read_only_completed=1
write_completed=1
approval_required=1
schema_rejected=1
unknown_action=1
authorization_denied=1
semantic_rejected=1
policy_blocked=1
idempotent_replay=1
idempotency_conflict=1
retryable_dependency_failures=1
nonretryable_dependency_failures=1
stale_state=1
response_validation_failed=1
bounded_retries=1
executed_side_effects=1
duplicate_side_effects=0
unauthorized_side_effects=0