Agents get assembled,
not authored.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import interrupt, Command

class FactoryState(TypedDict):
    request: dict
    route: str
    evidence: list[dict]
    draft: str
    risk: str          # low | medium | high | restricted
    steps: int

MAX_STEPS = 12
ROUTES = {"how_to": "knowledge", "access": "action", "outage": "triage"}

def supervisor(state: FactoryState):
    if state["steps"] >= MAX_STEPS:                 # every graph has a step budget
        return Command(goto=END, update={"draft": "Handing over — step budget reached."})
    intent = classify(state["request"]["utterance"])
    return Command(goto=ROUTES.get(intent, "knowledge"),
                   update={"route": intent, "steps": state["steps"] + 1})

def approval_gate(state: FactoryState):
    if state["risk"] in ("high", "restricted"):
        # execution pauses here and resumes on the approver's verdict
        verdict = interrupt({"summary": state["draft"], "risk": state["risk"]})
        if verdict["decision"] != "approve":
            return Command(goto=END, update={"draft": "Declined by approver."})
    return Command(goto="act")

builder = StateGraph(FactoryState)
builder.add_node("supervisor", supervisor)
builder.add_node("knowledge", knowledge_worker)
builder.add_node("triage", triage_worker)
builder.add_node("action", action_planner)
builder.add_node("approval_gate", approval_gate)
builder.add_node("act", action_executor)
builder.add_edge(START, "supervisor")
builder.add_edge("knowledge", END)
builder.add_edge("triage", "supervisor")
builder.add_edge("action", "approval_gate")
builder.add_edge("act", END)

graph = builder.compile(checkpointer=PostgresSaver.from_conn_string(PG_DSN))

# Tools are platform assets. Registered once, consumed by many agents.
TOOL_REGISTRY = {
    "itsm.reset_password": {
        "server": "https://mcp.internal/itsm",     # MCP is the standard tool interface
        "version": "2.3.0",
        "write": True,
        "risk": "high",
        "scopes": ["Incident.Write", "Identity.Reset"],
        "allowed_agents": ["itsm-resolver", "service-desk-copilot"],
        "max_calls_per_run": 1,
        "owner": "platform-integrations",
    },
    "kb.search": {
        "server": "https://mcp.internal/knowledge",
        "version": "1.8.2", "write": False, "risk": "low",
        "scopes": ["Knowledge.Read"], "allowed_agents": ["*"], "max_calls_per_run": 6,
        "owner": "platform-knowledge",
    },
}

class PolicyError(Exception):
    """Raised before a tool runs, never after."""

def authorize(agent_id: str, tool: str, ctx) -> dict:
    spec = TOOL_REGISTRY.get(tool)
    if spec is None:
        raise PolicyError(tool + " is not a registered tool")
    if "*" not in spec["allowed_agents"] and agent_id not in spec["allowed_agents"]:
        raise PolicyError(agent_id + " is not entitled to " + tool)
    if not set(spec["scopes"]).issubset(ctx.granted_scopes):
        raise PolicyError("user lacks scopes for " + tool)
    if spec["write"] and spec["risk"] == "high" and not ctx.human_approved:
        raise PolicyError("high-risk write requires recorded approval")
    if ctx.calls_made(tool) >= spec["max_calls_per_run"]:
        raise PolicyError("call budget exhausted for " + tool)
    return spec

async def invoke(agent_id: str, tool: str, args: dict, ctx):
    spec = authorize(agent_id, tool, ctx)
    with span("tool.call", tool=tool, version=spec["version"], agent=agent_id):
        return await mcp_client(spec["server"]).call(tool, args, headers=ctx.obo_headers)

# Routing policy is configuration. Squads call complete(), never a provider SDK.
ROUTES = {
    "restricted": {"model": "gpt-4o-eu",      "region": "westeurope", "fallback": None},
    "reasoning":  {"model": "claude-sonnet",  "fallback": "gpt-4o"},
    "standard":   {"model": "gpt-4o",         "fallback": "claude-sonnet"},
    "bulk":       {"model": "phi-4",          "fallback": "gpt-4o-mini"},
}
PRICE = {  # usd per 1M tokens: (input, output)
    "claude-sonnet": (3.00, 15.00), "gpt-4o": (2.50, 10.00),
    "gpt-4o-mini": (0.15, 0.60),    "phi-4": (0.07, 0.28),
}

def meter(model: str, usage) -> float:
    pin, pout = PRICE[model]
    return (usage.input_tokens * pin + usage.output_tokens * pout) / 1_000_000

async def complete(*, messages, tier, tenant, agent_id, sensitivity="standard"):
    route = ROUTES["restricted"] if sensitivity == "restricted" else ROUTES[tier]
    budget.assert_headroom(tenant, agent_id)          # stop runaway spend early
    for attempt, model in enumerate([route["model"], route.get("fallback")]):
        if not model:
            break
        try:
            resp = await provider_call(model, messages, timeout=route.get("timeout", 30))
        except (RateLimited, Timeout, ProviderDown) as err:
            emit_event("model.failover", model=model, reason=type(err).__name__)
            continue
        cost = meter(model, resp.usage)
        budget.charge(tenant, agent_id, cost)         # FinOps chargeback at source
        emit_span("model.call", model=model, degraded=bool(attempt),
                  tokens=resp.usage.total, cost_usd=cost)
        return resp
    raise AllModelsUnavailable(tier)

# Retrieval as a service. Security trimming happens in the query, not after it.
MIN_SCORE = 0.35
MIN_GROUNDEDNESS = 0.75

def acl_filter(user_upn: str, tenant: str) -> str:
    principals = identity.groups_for(user_upn) + ["upn:" + user_upn]
    allowed = ",".join(principals)
    return "tenant eq '" + tenant + "' and acl/any(g: search.in(g, '" + allowed + "'))"

def retrieve(query: str, user_upn: str, tenant: str, k: int = 8) -> list[dict]:
    candidates = search.hybrid(
        query_text=query,                 # BM25 leg: catches product codes and acronyms
        vector=embed(query),              # semantic leg
        filter=acl_filter(user_upn, tenant),
        top=40,
    )
    ranked = reranker.rank(query, candidates)[:k]
    fresh = [c for c in ranked if c.score >= MIN_SCORE and not c.is_expired]
    emit_span("rag.retrieve", candidates=len(candidates), returned=len(fresh))
    return fresh

def answer(query: str, ctx) -> AgentReply:
    chunks = retrieve(query, ctx.user_upn, ctx.tenant)
    if not chunks:
        return AgentReply(text="No source you have access to covers this yet.",
                          needs_human=True)
    reply = generate(query, chunks)                       # cite-or-abstain prompt
    score = groundedness(reply.text, chunks)              # claim-level check, 0..1
    emit_metric("rag.groundedness", score)
    if score < MIN_GROUNDEDNESS:
        return AgentReply(text="I could not ground this in approved sources.",
                          needs_human=True)
    return reply

-- Three gold assets every agent factory runs on.
CREATE CATALOG IF NOT EXISTS ai_factory;
CREATE SCHEMA  IF NOT EXISTS ai_factory.gold;

-- 1. Grounding: curated, permission-tagged, freshness-aware
CREATE TABLE ai_factory.gold.knowledge_chunk (
  chunk_id        STRING NOT NULL,
  tenant_id       STRING NOT NULL,
  source_system   STRING,          -- sharepoint | confluence | itsm_kb
  source_url      STRING,
  acl             ARRAY<STRING>,   -- group + upn principals, synced from source
  content         STRING,
  embedding       ARRAY<FLOAT>,
  valid_until     DATE,            -- expiry drives the freshness filter at query time
  ingested_at     TIMESTAMP
) USING DELTA PARTITIONED BY (tenant_id);

-- 2. Golden evaluation set: the release gate's source of truth
CREATE TABLE ai_factory.gold.eval_case (
  case_id      STRING, agent_id STRING,
  category     STRING,   -- happy | adversarial | refusal | acl_leak | ambiguity
  prompt       STRING, expected STRING, must_refuse BOOLEAN,
  as_user      STRING,   -- ACL cases run as a low-privilege identity
  added_by     STRING, added_at TIMESTAMP
) USING DELTA;

-- 3. Production telemetry: the improvement loop
CREATE TABLE ai_factory.gold.agent_trace (
  trace_id STRING, agent_id STRING, tenant_id STRING, pattern STRING,
  model STRING, tokens_in BIGINT, tokens_out BIGINT, cost_usd DECIMAL(10,5),
  groundedness FLOAT, tool_calls INT, escalated BOOLEAN,
  outcome STRING,       -- resolved | escalated | declined | failed
  latency_ms INT, ts TIMESTAMP
) USING DELTA PARTITIONED BY (agent_id);

-- Weekly quality review, straight off the lakehouse
SELECT agent_id, pattern,
       COUNT(*)                                        AS runs,
       AVG(CASE WHEN outcome = 'resolved' THEN 1 ELSE 0 END) AS autonomy_rate,
       AVG(groundedness)                               AS mean_groundedness,
       SUM(cost_usd) / SUM(CASE WHEN outcome = 'resolved' THEN 1 ELSE 0 END)
                                                       AS cost_per_resolution
FROM   ai_factory.gold.agent_trace
WHERE  ts >= current_date() - INTERVAL 7 DAYS
GROUP  BY agent_id, pattern
ORDER  BY cost_per_resolution DESC;

import hashlib

def idempotency_key(thread_id: str, action: str, subject: str) -> str:
    return hashlib.sha256("|".join([thread_id, action, subject]).encode()).hexdigest()

@action(name="itsm.reset_password",
        compensates="itsm.revoke_temp_credential",
        risk="high")
async def reset_password(req, ctx):
    key = idempotency_key(ctx.thread_id, "itsm.reset_password", req.user_upn)

    prior = ledger.get(key)
    if prior and prior.state == "committed":
        return prior.result                      # replay-safe: same call, same answer

    if ctx.dry_run:                              # evaluations and approvers use this path
        return Preview("Would reset the password for " + req.user_upn)

    ledger.begin(key, action="itsm.reset_password", actor=ctx.user_upn, payload=req.dict())
    try:
        result = await itsm.post("/identity/password/reset",
                                 json=req.dict(),
                                 headers=ctx.obo_headers)   # acts AS the user
    except Exception as err:
        ledger.fail(key, str(err))
        await compensate(key, ctx)               # unwind anything already applied
        raise
    ledger.commit(key, result)
    emit_event("action.committed", action="itsm.reset_password",
               actor=ctx.user_upn, trace_id=ctx.trace_id)
    return result