Multi-Agent Orchestration

from dataclasses import dataclass

@dataclass(frozen=True)
class TaskShape:
    independent_branches: int
    agentic_branches: int
    needs_full_shared_trace: bool
    merge_contract_defined: bool
    value_justifies_extra_cost: bool

def choose_orchestration(task: TaskShape) -> str:
    if not 0 <= task.agentic_branches <= task.independent_branches:
        raise ValueError("agentic branches must be a subset of branches")
    if (
        task.independent_branches >= 2
        and task.agentic_branches >= 2
        and not task.needs_full_shared_trace
        and task.merge_contract_defined
        and task.value_justifies_extra_cost
    ):
        return "parallel multi-agent graph"
    if (
        task.independent_branches >= 2
        and task.agentic_branches == 0
        and not task.needs_full_shared_trace
        and task.merge_contract_defined
    ):
        return "parallel deterministic workflow"
    if task.agentic_branches == 0:
        return "serial deterministic workflow"
    return "single scoped agent"

release_checks = TaskShape(3, 2, False, True, True)
deterministic_etl = TaskShape(3, 0, False, True, True)
coupled_code_edit = TaskShape(3, 3, True, False, True)
cheap_semantic_task = TaskShape(3, 2, False, True, False)
assert choose_orchestration(cheap_semantic_task) == "single scoped agent"
assert choose_orchestration(TaskShape(1, 0, False, True, False)) == "serial deterministic workflow"

print("release checks:", choose_orchestration(release_checks))
print("deterministic ETL:", choose_orchestration(deterministic_etl))
print("coupled code edit:", choose_orchestration(coupled_code_edit))
print("semantic task without budget:", choose_orchestration(cheap_semantic_task))

import asyncio
import hashlib
import json
import operator
from collections import Counter
from copy import deepcopy
from types import MappingProxyType
from typing import Annotated, NotRequired, TypedDict

from langgraph.graph import END, START, StateGraph

class Receipt(TypedDict):
    branch: str
    tenant_id: str
    request_id: str
    release_id: str
    release_version: int
    observed_at: int
    ok: bool
    evidence_ref: str
    source_digest: str

class ReleaseState(TypedDict):
    tenant_id: str
    request_id: str
    release_id: str
    release_version: int
    requested_traffic: int
    now: int
    receipts: Annotated[list[Receipt], operator.add]
    proposal: NotRequired[dict[str, object]]
    outcome: NotRequired[str]

initial: ReleaseState = {
    "tenant_id": "tenant-7", "request_id": "promote-104",
    "release_id": "reranker-v17", "release_version": 3,
    "requested_traffic": 25, "now": 3600, "receipts": [],
}
# Synthetic source records. These are not live metrics or model judgments.
SOURCES = {
    "evals/run-91": {"branch": "eval", "regressions": 2, "cases": 100},
    "metrics/window-1h": {"branch": "health", "errors": 4, "requests": 10000},
    "policy/rollout-v4": {"branch": "policy", "max_traffic": 25},
}
for record in SOURCES.values():
    record.update(tenant_id="tenant-7", request_id="promote-104",
                  release_id="reranker-v17", release_version=3, observed_at=3500)
REFS = {record["branch"]: ref for ref, record in SOURCES.items()}

def source_digest(record):
    # Digest this fixture's JSON-compatible source content, not just its location.
    return hashlib.sha256(json.dumps(record, sort_keys=True, separators=(",", ":")).encode()).hexdigest()

def source_passes(record, traffic):
    if record["branch"] == "eval":
        return record["cases"] > 0 and record["regressions"] / record["cases"] <= 0.03
    if record["branch"] == "health":
        # Fictional release rule: <= 1x burn against a 99.9% success SLO.
        return record["requests"] > 0 and (record["errors"] / record["requests"]) / 0.001 <= 1
    return traffic <= record["max_traffic"]

def receipt(state, branch, evidence_ref, *, ok=None, tenant_id=None):
    record = SOURCES[evidence_ref]
    return {
        "branch": branch,
        "tenant_id": record["tenant_id"] if tenant_id is None else tenant_id,
        "request_id": record["request_id"], "release_id": record["release_id"],
        "release_version": record["release_version"], "observed_at": record["observed_at"],
        "ok": source_passes(record, state["requested_traffic"]) if ok is None else ok,
        "evidence_ref": evidence_ref,
        "source_digest": source_digest(record),
    }

def checked_update(branch, update):
    if set(update) != {"receipts"} or not isinstance(update["receipts"], list):
        raise ValueError("branch may write only its receipt")
    if len(update["receipts"]) != 1 or update["receipts"][0]["branch"] != branch:
        raise ValueError("branch identity mismatch")
    return deepcopy(update)

def validate_target(state):
    traffic = state["requested_traffic"]
    if type(traffic) is not int or not 0 <= traffic <= 100:
        raise ValueError("traffic target must be an integer percentage in [0, 100]")

def merge_receipts(state):
    validate_target(state)
    receipts = state["receipts"]
    expected = {"eval", "health", "policy"}
    counts = Counter(item["branch"] for item in receipts)
    if len(receipts) != 3 or set(counts) != expected or any(n != 1 for n in counts.values()):
        raise ValueError("expected exactly one eval, health, and policy receipt")
    by_branch = {item["branch"]: item for item in receipts}
    for branch, item in by_branch.items():
        for field in ("tenant_id", "request_id", "release_id", "release_version"):
            if item[field] != state[field]:
                raise ValueError(f"{field} mismatch")
        if type(item["observed_at"]) is not int or not 0 <= state["now"] - item["observed_at"] <= 300:
            raise ValueError("stale or future evidence")
        if item["evidence_ref"] != REFS[branch]:
            raise ValueError("unrecognized evidence reference")
        # Resolve the reference and recompute the decision; a model's `ok` isn't proof.
        trusted = receipt(state, branch, REFS[branch])
        if item != trusted or item["ok"] is not True:
            raise ValueError(f"invalid {branch} receipt")
    action = {
        "kind": "promote", "tenant_id": state["tenant_id"],
        "release_id": state["release_id"], "release_version": state["release_version"],
        "traffic_percent": state["requested_traffic"],
    }
    evidence = [{"ref": by_branch[branch]["evidence_ref"],
                 "digest": by_branch[branch]["source_digest"]} for branch in sorted(expected)]
    packet = {"request_id": state["request_id"], "action": action, "evidence": evidence}
    digest = hashlib.sha256(json.dumps(packet, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
    return {"proposal": {**packet, "digest": digest, "idempotency_key": f"promotion:{digest}"},
            "outcome": "awaiting_approval"}

def build_read_graph():
    # This rendezvous proves overlap without asserting a machine-dependent runtime.
    started = set()
    all_started = asyncio.Event()

    def branch_node(branch):
        async def run(state):
            view = MappingProxyType({k: deepcopy(state[k]) for k in initial if k != "receipts"})
            started.add(branch)
            if len(started) == 3:
                all_started.set()
            await asyncio.wait_for(all_started.wait(), timeout=5)
            return checked_update(branch, {"receipts": [receipt(view, branch, REFS[branch])]})
        return run

    builder = StateGraph(ReleaseState)
    def triage(state):
        validate_target(state)  # TypedDict doesn't validate values at runtime.
        return {"outcome": "checking"}
    builder.add_node("triage", triage)
    builder.add_edge(START, "triage")
    for branch in ("eval", "health", "policy"):
        builder.add_node(branch, branch_node(branch))
        builder.add_edge("triage", branch)
    builder.add_node("merge", merge_receipts)
    builder.add_edge(["eval", "health", "policy"], "merge")
    builder.add_edge("merge", END)
    return builder.compile()

state = asyncio.run(build_read_graph().ainvoke(initial, {"max_concurrency": 3}))
assert sorted(item["branch"] for item in state["receipts"]) == ["eval", "health", "policy"]
assert state["outcome"] == "awaiting_approval"
assert "proposal" not in initial and initial["receipts"] == []
print("three read branches overlapped; one receipt per branch")
print("eval regression rate: 2.0%; canary burn: 0.4x; policy ceiling: 25%")
print("outcome:", state["outcome"])
print("writer nodes in graph:", 0)

import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.errors import InvalidUpdateError

class State(TypedDict):
    receipts: Annotated[list[str], operator.add]
    merges: Annotated[list[list[str]], operator.add]

def run(wait_for_all, *, select_health=True, defer=False):
    b = StateGraph(State)
    for name in ("eval", "health", "policy"):
        b.add_node(name, lambda s, name=name: {"receipts": [name]})
    b.add_node("prepare_health", lambda s: {})
    b.add_node("merge", lambda s: {"merges": [sorted(s["receipts"])]}, defer=defer)
    selected = ("eval", "prepare_health", "policy") if select_health else ("eval", "policy")
    for name in selected:
        b.add_edge(START, name)
    b.add_edge("prepare_health", "health")
    if wait_for_all:
        b.add_edge(["eval", "health", "policy"], "merge")
    else:
        for name in ("eval", "health", "policy"):
            b.add_edge(name, "merge")
    b.add_edge("merge", END)
    return b.compile().invoke({"receipts": [], "merges": []})["merges"]

joined = run(True)
separate = run(False)
assert joined == [["eval", "health", "policy"]]
assert separate == [["eval", "policy"], ["eval", "health", "policy"]]
print("waiting edge:", joined)
print("separate edges:", separate)
skipped = run(True, select_health=False)
selected = run(False, select_health=False, defer=True)
assert skipped == [] and selected == [["eval", "policy"]]
print("waiting for skipped health:", skipped)
print("deferred selected branches:", selected)

class Scalar(TypedDict):
    traffic: int
b = StateGraph(Scalar)
for name, value in (("a", 10), ("b", 25)):
    b.add_node(name, lambda s, value=value: {"traffic": value})
    b.add_edge(START, name)
    b.add_edge(name, END)
try:
    b.compile().invoke({"traffic": 0})
except InvalidUpdateError:
    print("unreduced concurrent scalar: rejected by LangGraph")
else:
    raise AssertionError("ambiguous scalar writes should fail")

from copy import deepcopy
from types import MappingProxyType

source = {"receipts": []}
view = MappingProxyType(source)
view["receipts"].append("unexpected")
assert source["receipts"] == ["unexpected"]
print("proxy nested mutation reached source:", True)

copied = MappingProxyType(deepcopy(source))
copied["receipts"].append("local change")
assert source["receipts"] == ["unexpected"]
print("copied view nested mutation reached source:", False)

from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt

class State(TypedDict):
    digest: str
    outcome: str

entries = []
def review(state):
    entries.append(state["digest"])  # Local instrumentation, not an external effect.
    decision = interrupt({"question": "Approve this packet?", "digest": state["digest"]})
    approved = (isinstance(decision, dict) and decision.get("approved") is True
                and decision.get("digest") == state["digest"])
    return Command(goto="approved" if approved else "cancelled")

b = StateGraph(State)
b.add_node("review", review)
b.add_node("approved", lambda s: {"outcome": "approved-for-writer-checks"})
b.add_node("cancelled", lambda s: {"outcome": "cancelled"})
b.add_edge(START, "review")
b.add_edge("approved", END)
b.add_edge("cancelled", END)
graph = b.compile(checkpointer=InMemorySaver())
for i, decision in enumerate((
    {"approved": True, "digest": "packet-104"},
    {"approved": False, "digest": "packet-104"},
    {"approved": True, "digest": "different-packet"},
    "approve everything",
)):
    config = {"configurable": {"thread_id": f"tenant-7:review-{i}"}}
    paused = graph.invoke({"digest": "packet-104", "outcome": "pending"}, config)
    assert paused["__interrupt__"] and paused["outcome"] == "pending"
    result = graph.invoke(Command(resume=decision), config)
    expected = "approved-for-writer-checks" if i == 0 else "cancelled"
    assert result["outcome"] == expected
    print(f"decision {i}:", result["outcome"])
assert len(entries) == 8  # Each interrupted node entered twice.
print("node entries:", len(entries), "for 4 reviews")

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy

class State(TypedDict):
    value: int

def exercise(mode):
    calls = []
    def read(state):
        calls.append(1)
        if mode == "permanent":
            raise ValueError("unknown release")
        if mode == "exhausted" or len(calls) < 3:
            raise TimeoutError("metrics unavailable")
        return {"value": 4}
    b = StateGraph(State)
    b.add_node("read", read, retry_policy=RetryPolicy(
        max_attempts=3, initial_interval=0, jitter=False, retry_on=TimeoutError))
    b.add_edge(START, "read")
    b.add_edge("read", END)
    try:
        result = b.compile().invoke({"value": 0})
    except (ValueError, TimeoutError):
        assert mode != "transient"
    else:
        assert mode == "transient" and result["value"] == 4
    return len(calls)

assert exercise("transient") == 3
assert exercise("permanent") == 1
assert exercise("exhausted") == 3
print("attempts: transient=3, permanent=1, exhausted=3")

import hashlib
import json
from dataclasses import asdict, dataclass, replace

@dataclass(frozen=True)
class Action:
    tenant: str
    request: str
    release: str
    version: int
    traffic: int
    policy_version: int
    evidence_revision: str
    valid_until: int

def digest(action):
    return hashlib.sha256(json.dumps(asdict(action), sort_keys=True).encode()).hexdigest()

# These maps stand in for trusted server-side state, not model output.
writer_tenants = {"writer-service": {"tenant-7"}}
reviewer_tenants = {"reviewer-9": {"tenant-7"}}
live = {("tenant-7", "reranker-v17"): {"version": 3, "traffic": 10, "policy_version": 4}}
approvals = {}  # digest -> authenticated reviewer and expiry
ledger = {}     # (tenant, idempotency_key) -> (action, volatile fixture result)

def approve(action, reviewer, now):
    if action.tenant not in reviewer_tenants.get(reviewer, set()) or now >= action.valid_until:
        raise PermissionError("reviewer scope or expired evidence")
    approvals[digest(action)] = (reviewer, action.valid_until)

def execute(action, actor, key, now):
    if action.tenant not in writer_tenants.get(actor, set()):
        raise PermissionError("writer scope")
    if type(action.traffic) is not int or not 0 <= action.traffic <= 100:
        raise ValueError("invalid traffic percentage")
    ledger_key = (action.tenant, key)
    if ledger_key in ledger:
        original, result = ledger[ledger_key]
        if original != action:
            raise ValueError("idempotency key reused for different action")
        return "reused:" + result  # Reconcile a prior result, not a new write.
    if key != "promotion:" + digest(action):
        raise ValueError("key doesn't bind this proposal")
    approval = approvals.get(digest(action))
    if approval is None:
        raise PermissionError("no approval for exact proposal")
    reviewer, expires = approval
    if action.tenant not in reviewer_tenants.get(reviewer, set()) or now >= expires:
        raise PermissionError("approval expired or reviewer revoked")
    current = live.get((action.tenant, action.release))
    if current is None or (current["version"], current["policy_version"]) != (action.version, action.policy_version):
        raise ValueError("live release or policy changed")
    # One atomic transaction in production: compare versions, mutate, record result.
    current.update(traffic=action.traffic, version=action.version + 1)
    result = f"promotion-{len(ledger) + 1}"
    ledger[ledger_key] = (action, result)
    return "created:" + result

action = Action("tenant-7", "promote-104", "reranker-v17", 3, 25, 4, "eval91-health3500-policy4", 120)
key = "promotion:" + digest(action)
def denied(label, operation):
    before = (repr(live), len(ledger))
    try:
        operation()
    except (ValueError, PermissionError):
        assert (repr(live), len(ledger)) == before
        print(label + ": blocked")
    else:
        raise AssertionError(label)

denied("unapproved", lambda: execute(action, "writer-service", key, 100))
approve(action, "reviewer-9", 100)
denied("writer can't approve", lambda: approve(action, "writer-service", 100))
denied("worker write", lambda: execute(action, "health-agent", key, 100))
other = replace(action, tenant="tenant-8")
denied("cross-tenant", lambda: execute(other, "writer-service", "promotion:" + digest(other), 100))
denied("expired", lambda: execute(action, "writer-service", key, 120))
reviewer_tenants["reviewer-9"] = set()
denied("reviewer revoked", lambda: execute(action, "writer-service", key, 100))
reviewer_tenants["reviewer-9"] = {"tenant-7"}
live[(action.tenant, action.release)]["policy_version"] = 5
denied("policy changed", lambda: execute(action, "writer-service", key, 100))
live[(action.tenant, action.release)]["policy_version"] = 4
live[(action.tenant, action.release)]["version"] = 4
denied("stale version", lambda: execute(action, "writer-service", key, 100))
live[(action.tenant, action.release)]["version"] = 3
print("first:", execute(action, "writer-service", key, 100))
# Simulate a lost response: replay after the successful write advanced version.
print("replay:", execute(action, "writer-service", key, 101))
denied("changed action", lambda: execute(replace(action, traffic=50), "writer-service", key, 101))
assert len(ledger) == 1 and live[(action.tenant, action.release)]["traffic"] == 25
print("ledger records:", len(ledger))

from agents import Agent

release_agent = Agent(
    name="Release Agent",
    handoff_description="Handles release status and rollout-plan questions.",
    instructions="Explain release evidence. Never execute a traffic shift.",
)
health_agent = Agent(
    name="Health Agent",
    handoff_description="Handles canary metrics and error-budget questions.",
    instructions="Explain current health evidence from approved tools.",
)
triage_agent = Agent(
    name="Triage Agent",
    instructions="Transfer this conversation to one matching specialist.",
    handoffs=[release_agent, health_agent],
)

def try_merge(label: str, receipts: list[Receipt]) -> None:
    probe = dict(initial)
    probe["receipts"] = receipts
    try:
        merge_receipts(probe)
        raise AssertionError(f"{label}: unexpectedly accepted")
    except ValueError as exc:
        print(f"{label}: {exc}")

try_merge(
    "health fail",
    [
        receipt(initial, "eval", "evals/run-91"),
        receipt(initial, "health", "metrics/window-1h", ok=False),
        receipt(initial, "policy", "policy/rollout-v4"),
    ],
)
try_merge(
    "tenant mismatch",
    [
        receipt(initial, "eval", "evals/run-91"),
        receipt(initial, "health", "metrics/window-1h"),
        receipt(initial, "policy", "policy/rollout-v4", tenant_id="tenant-8"),
    ],
)
try_merge(
    "incomplete",
    [
        receipt(initial, "eval", "evals/run-91"),
        receipt(initial, "health", "metrics/window-1h"),
    ],
)
try_merge(
    "duplicate policy",
    [
        receipt(initial, "eval", "evals/run-91"),
        receipt(initial, "health", "metrics/window-1h"),
        receipt(initial, "policy", "policy/rollout-v4"),
        receipt(initial, "policy", "policy/rollout-v4"),
    ],
)
try_merge(
    "unexpected branch",
    [
        receipt(initial, "eval", "evals/run-91"),
        receipt(initial, "health", "metrics/window-1h"),
        receipt(initial, "audit", "policy/rollout-v4"),
    ],
)

good = [receipt(initial, branch, REFS[branch]) for branch in ("eval", "health", "policy")]
for bad in (-1, 101, True, "25", float("nan")):
    try:
        merge_receipts({**initial, "requested_traffic": bad, "receipts": good})
    except ValueError as exc:
        assert str(exc) == "traffic target must be an integer percentage in [0, 100]"
    else:
        raise AssertionError("invalid target became a proposal")
print("invalid traffic targets blocked:", 5)
for label, change in (
    ("wrong release", {"release_id": "reranker-v16"}),
    ("old release version", {"release_version": 2}),
    ("stale evidence", {"observed_at": 3200}),
    ("future evidence", {"observed_at": 3601}),
    ("invented source", {"evidence_ref": "evals/invented"}),
    ("string success", {"ok": "true"}),
):
    try_merge(label, [{**good[0], **change}, *good[1:]])
for label, update in (
    ("scalar overwrite", {"receipts": [good[0]], "requested_traffic": 100}),
    ("branch spoof", {"receipts": [good[1]]}),
):
    try:
        checked_update("eval", update)
    except ValueError:
        print(label + ": blocked")
    else:
        raise AssertionError(label)

# The location stays identical while the source content changes.
original_packet = merge_receipts({**initial, "receipts": good})["proposal"]
SOURCES["evals/run-91"]["regressions"] = 1
try_merge("source changed at same reference", good)
fresh = [receipt(initial, branch, REFS[branch]) for branch in ("eval", "health", "policy")]
new_packet = merge_receipts({**initial, "receipts": fresh})["proposal"]
assert new_packet["digest"] != original_packet["digest"]
print("new source content changes proposal digest:", True)
SOURCES["evals/run-91"]["regressions"] = 2