johnandersen777.bsky.social/atlibs-asyncai

Switch repository

Loading…

//! The asyncai agent loop (github.com/publicdomainrelay/asyncai openai_backend.rs) on the step
//! protocol: MCP servers are initialized and their tools listed, then inference (the Anthropic
//! Messages API, or OpenAI chat completions: co/core and any compatible endpoint) -> tool_use ->
//! MCP tools/call -> tool_result -> inference ... until the model stops or the step cap. One MCP client, two transports:
//! Streamable HTTP through the host's com.fedfork.exec.http.fetch (https, or any scheme the host fetches: webrtc://,
//! iroh://), or a composition hook (one JSON-RPC message in, its response out). With dynamicMcp the model connects
//! and disconnects servers mid-run (the built-in tools mcp_connect and mcp_disconnect).

use abi::{Call, Event, Handle, Run};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};

abi::entries! { run => run_handler }

const HTTP: &str = "com.fedfork.exec.http.fetch";
/// The transcript lives in the host (atlibs held values): appended to, spliced into each inference
/// body and into the result, so it never rides in the state or passes back through the module.
const APPEND: &str = "com.fedfork.exec.held.append";
/// A compaction replaces the held transcript (then the kept messages are appended after it).
const PUT: &str = "com.fedfork.exec.held.put";
const TRANSCRIPT: &str = "messages";
/// OpenAI chat completions: the system message, held apart and sent before the transcript.
const SYSTEM: &str = "system";
/// provider openai's default endpoint: co/core (https://cocore.dev/docs/inference)
const OPENAI_ENDPOINT: &str = "https://cocore.dev/api/v1";
/// The in-process MCP hook: input one JSON-RPC message, output its response (null for a notification).
const MCP_HOOK: &str = "com.fedfork.exec.mcp.message";
/// An MCP request's deadline (ms), past any waitMs/timeoutMs its tool call carries (at most MCP_WAIT_MAX of it): the
/// guest agent's exec runs 60 s by default and up to 10 min, its read long-polls up to 30 s. A dead transport then
/// fails the call instead of holding the run to its wallMs.
const MCP_TIMEOUT: u64 = 120_000;
const MCP_WAIT_MAX: u64 = 600_000;
/// What initialize asks for, and the versions this client speaks (Streamable HTTP: 2025-03-26 on).
const PROTOCOL: &str = "2025-11-25";
const SUPPORTED: [&str; 3] = ["2025-11-25", "2025-06-18", "2025-03-26"];
const MODEL: &str = "claude-opus-5-5";
const ENDPOINT: &str = "https://api.anthropic.com";
const ANTHROPIC_VERSION: &str = "2026-10-21";
const MAX_TOKENS: u64 = 16_000;
/// Inference calls per run; the last one may not use tools (asyncai's MAX_TOOL_ROUNDS + 1).
const MAX_STEPS: u64 = 9;
const RETRIES: u64 = 3;
const LIST_PAGES: u64 = 50;
/// The built-in tools (dynamicMcp): connect a server mid-run, disconnect one.
const CONNECT: &str = "mcp_connect";
const DISCONNECT: &str = "mcp_disconnect";
/// Servers a run may have, the disconnected and failed ones included (each connect adds one).
const MAX_SERVERS: usize = 32;
/// Compaction: past compactAt of contextWindow (tokens) the transcript is summarized.
const CONTEXT_WINDOW: u64 = 200_000;
const COMPACT_AT: f64 = 0.8;
/// The server-side compaction beta (Anthropic): its trigger is at least this many input tokens.
const SERVER_COMPACT_BETA: &str = "compact-2026-01-12";
const SERVER_COMPACT_MIN: u64 = 50_000;
/// The summary request's instruction (after Anthropic's default compaction prompt), added to the last user message.
const COMPACT_PROMPT: &str = "You have written a partial transcript for the initial task above. Please write a summary of the transcript. \
The purpose of this summary is to provide continuity so you can continue to make progress towards solving the task in a future context, \
where the raw history above may not be accessible and will be replaced with this summary. Write down anything that would be helpful, \
including the state, next steps, learnings, the user's requests and any exact values (names, ids, numbers) you will need. \
You must wrap your summary in a <summary></summary> block. Do not call any tools: answer with the summary only.";
/// The first message of a compacted transcript starts with this (a page can show it as the compaction).
const SUMMARY_PREFIX: &str = "This session continues an earlier conversation that was compacted to save context. Summary of the earlier conversation:";
const TOOL_BUDGET_EXHAUSTED: &str = "Tool budget exhausted: no more tool calls are possible. \
Answer now using only the results you already have, and say briefly what you could not determine.";

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct Args {
    prompt: Option<String>,
    system: Option<String>,
    model: Option<String>,
    endpoint: Option<String>,
    /// "anthropic" (the Messages API, the default) or "openai" (chat completions)
    provider: Option<String>,
    /// openai: ask for an event stream (collected whole) instead of one JSON answer
    stream: Option<Value>,
    max_tokens: Option<Value>,
    max_steps: Option<Value>,
    /// [{ name, url } | { name, hook? } | { name, server: <one of those> }], an array or its JSON (a form field).
    mcp: Option<Value>,
    /// mcp_connect / mcp_disconnect offered: true (url descriptors), or a list of "url", URL prefixes and hook NSIDs
    dynamic_mcp: Option<Value>,
    /// Fields merged over the Messages request body (thinking, output_config, fallbacks, ...).
    extra: Option<Value>,
    /// "client" (the default): summarize with an inference call; "server": Anthropic's compaction beta; "off" (or false)
    compact: Option<Value>,
    context_window: Option<Value>,
    /// the fraction of contextWindow that triggers a compaction
    compact_at: Option<Value>,
    /// replaces COMPACT_PROMPT (and is the server's instructions)
    compact_prompt: Option<String>,
}

#[derive(Serialize, Deserialize, Clone, Default)]
struct Server {
    name: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    hook: Option<String>,
    /// connected by the model (mcp_connect)
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    dynamic: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    session: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    version: Option<String>,
    /// init | initialized | list | ready | skipped | error | disconnected (mcp_disconnect, or an mcp_connect that failed)
    phase: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    error: Option<String>,
    /// tools/list as the server answered it
    #[serde(default)]
    listed: Vec<Value>,
    /// tool names left out (a name past 64 characters, or a clash)
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    dropped: Vec<String>,
}

#[derive(Serialize, Deserialize, Default)]
struct State {
    /// messages not yet appended to the held transcript (they go with the next inference or the finish)
    outbox: Vec<Value>,
    /// the text blocks of the last assistant message
    text: Vec<String>,
    servers: Vec<Server>,
    /// The Messages API tools: namespaced <server>__<tool>
    tools: Vec<Value>,
    /// namespaced name -> [server index, MCP tool name]
    routes: Map<String, Value>,
    steps: u64,
    retries: u64,
    /// this tool_use turn's calls: (call id, tool_use id, server index, MCP tool name, slot in results; an
    /// mcp_connect's call id is connect.<server index>, answered when that server is ready or failed),
    /// and its tool_result blocks in tool_use order (null until answered)
    waiting: Vec<(String, String, usize, String, usize)>,
    results: Vec<Value>,
    usage: Map<String, Value>,
    next: u64,
    /// the last inference's input + output tokens: the transcript's size when the next request goes
    context: u64,
    compactions: u64,
    /// a compaction's summary request is out: the messages it keeps (from the last assistant turn on)
    #[serde(default, skip_serializing_if = "Option::is_none")]
    kept: Option<Vec<Value>>,
    /// the held transcript's new start, put before the outbox is next appended (a compaction)
    #[serde(default, skip_serializing_if = "Option::is_none")]
    base: Option<Vec<Value>>,
    /// just compacted: the next request goes without checking again
    fresh: bool,
    /// openai: the system message is held (the first append puts it)
    #[serde(default)]
    started: bool,
    /// the start's held.append answered the earlier turns' size
    #[serde(default)]
    sized: bool,
}

fn run_handler(run: Run<Args>) -> Result<Handle, String> {
    let args = run.args;
    let config = Config::of(&args)?;
    let mut state: State = abi::state(run.state)?;
    match run.event {
        Event::Start => {
            state.servers = servers(&args)?;
            let prompt = args.prompt.as_deref().unwrap_or_default();
            if prompt.trim().is_empty() {
                return Err("prompt is required".into());
            }
            // earlier turns, when any, are the host's held `messages` already: the prompt is appended to them
            state.outbox = vec![json!({ "role": "user", "content": prompt })];
            // the transcript's size so far (the earlier turns the host holds): an empty append answers it
            let size = Call::new("held.start", APPEND).input(json!({ "name": TRANSCRIPT, "items": [] }));
            let calls: Vec<Call> = std::iter::once(size).chain(state.servers.iter().enumerate().map(|(i, s)| initialize(i, s))).collect();
            Ok(calls.into_iter().fold(Handle::new(), Handle::call).state(json!(state)))
        }
        Event::Result { id, result, .. } => {
            let output = result.into_result();
            let (kind, rest) = id.split_once('.').unwrap_or((&id, ""));
            let index = || rest.split('.').next().and_then(|i| i.parse::<usize>().ok()).filter(|i| *i < state.servers.len());
            match kind {
                "init" => {
                    let i = index().ok_or("bad call id")?;
                    let server = &mut state.servers[i];
                    match output.and_then(|o| reply(server, &id, &o).map(|r| (r, o))) {
                        Ok((None, _)) => server.phase = "skipped".into(),
                        Ok((Some(r), o)) => match r["protocolVersion"].as_str().filter(|v| SUPPORTED.contains(v)) {
                            Some(v) => {
                                server.version = Some(v.into());
                                server.session = header(&o, "mcp-session-id").map(String::from);
                                server.phase = "initialized".into();
                                let call = mcp_call(&format!("inited.{i}"), server, json!({ "jsonrpc": "2.0", "method": "notifications/initialized" }));
                                return Ok(Handle::new().call(call).state(json!(state)));
                            }
                            None => fail(server, format!("unsupported protocolVersion {}", r["protocolVersion"])),
                        },
                        Err(e) => fail(server, e),
                    }
                    Ok(ready(state, &config))
                }
                "inited" => {
                    let i = index().ok_or("bad call id")?;
                    let server = &mut state.servers[i];
                    match output.and_then(|o| notified(server, &o)) {
                        Ok(()) => {
                            server.phase = "list".into();
                            let call = list(i, 0, server, None);
                            Ok(Handle::new().call(call).state(json!(state)))
                        }
                        Err(e) => {
                            fail(server, e);
                            Ok(ready(state, &config))
                        }
                    }
                }
                "list" => {
                    let i = index().ok_or("bad call id")?;
                    let page: u64 = rest.split('.').nth(1).and_then(|p| p.parse().ok()).unwrap_or(0);
                    let server = &mut state.servers[i];
                    match output.and_then(|o| reply(server, &id, &o)).and_then(|r| r.ok_or_else(|| "no response to tools/list".to_string())) {
                        Ok(r) => {
                            server.listed.extend(r["tools"].as_array().cloned().unwrap_or_default());
                            match r["nextCursor"].as_str().filter(|c| !c.is_empty() && page + 1 < LIST_PAGES) {
                                Some(cursor) => {
                                    let call = list(i, page + 1, server, Some(cursor));
                                    return Ok(Handle::new().call(call).state(json!(state)));
                                }
                                None => server.phase = "ready".into(),
                            }
                        }
                        Err(e) => fail(server, e),
                    }
                    Ok(ready(state, &config))
                }
                "infer" => Ok(inferred(state, &config, output)),
                "compact" => Ok(compacted(state, &config, output)),
                // the transcript past the host's heldBytes
                "held" => Ok(match output {
                    Err(e) => finish(state, &config, "error", Some(format!("transcript: {e}"))),
                    // about 4 bytes a token, until an answer's usage says
                    Ok(o) if rest == "start" => {
                        state.context = o["bytes"].as_u64().unwrap_or(0) / 4;
                        state.sized = true;
                        ready(state, &config)
                    }
                    Ok(_) => Handle::new(),
                }),
                "tool" => {
                    let Some((_, use_id, i, name, k)) = state.waiting.iter().find(|w| w.0 == id).cloned() else {
                        return Ok(Handle::new());
                    };
                    let content = output.and_then(|o| reply(&state.servers[i], &id, &o)).and_then(|r| r.ok_or_else(|| "no response to tools/call".to_string()));
                    state.results[k] = tool_result(&config, &use_id, match content {
                        Ok(r) => tool_text(&r),
                        Err(e) => (format!("tool error: {name}: {e}"), true),
                    });
                    Ok(answered(state, &config))
                }
                _ => Ok(Handle::new()),
            }
        }
        Event::Timer { id } if id == "retry" => Ok(infer(state, &config)),
        Event::Timer { id } if id == "recompact" => {
            let call = summary_call(&state, &config);
            Ok(Handle::new().call(call).state(json!(state)))
        }
        _ => Ok(Handle::new()),
    }
}

#[derive(PartialEq, Clone, Copy)]
enum Provider {
    Anthropic,
    OpenAi,
}

#[derive(PartialEq)]
enum Compact {
    Off,
    Client,
    Server,
}

struct Config {
    provider: Provider,
    stream: bool,
    prompt: String,
    compact: Compact,
    /// tokens: past this the transcript is compacted
    threshold: u64,
    compact_prompt: Option<String>,
    model: String,
    url: String,
    max_tokens: u64,
    max_steps: u64,
    system: Option<String>,
    extra: Map<String, Value>,
    /// what mcp_connect may connect: "url" (any URL), URL prefixes (any scheme: webrtc:/// is every webrtc URL), hook NSIDs; empty: off
    dynamic: Vec<String>,
}

impl Config {
    fn of(args: &Args) -> Result<Self, String> {
        let provider = match args.provider.as_deref().map(str::trim).unwrap_or_default() {
            "" | "anthropic" => Provider::Anthropic,
            "openai" => Provider::OpenAi,
            other => return Err(format!("provider {other}: anthropic or openai")),
        };
        let default = if provider == Provider::OpenAi { OPENAI_ENDPOINT } else { ENDPOINT };
        let endpoint = args.endpoint.as_deref().map(str::trim).filter(|e| !e.is_empty()).unwrap_or(default).trim_end_matches('/');
        let stream = matches!(opt_json_or_text(&args.stream), Some(Value::Bool(true))) || opt_json_or_text(&args.stream) == Some(json!("true"));
        if stream && provider != Provider::OpenAi {
            return Err("stream is for provider openai".into());
        }
        let model = args.model.clone().filter(|m| !m.trim().is_empty());
        if model.is_none() && provider == Provider::OpenAi {
            return Err("model is required with provider openai".into());
        }
        let extra = match opt_json(&args.extra, "extra")? {
            None => Map::new(),
            Some(Value::Object(m)) => m,
            Some(_) => return Err("extra must be an object".into()),
        };
        let compact = match opt_json_or_text(&args.compact) {
            None => Compact::Client,
            Some(Value::Bool(false)) => Compact::Off,
            Some(Value::Bool(true)) => Compact::Client,
            Some(Value::String(s)) if s == "client" => Compact::Client,
            Some(Value::String(s)) if s == "server" => Compact::Server,
            Some(Value::String(s)) if s == "off" || s == "false" => Compact::Off,
            Some(_) => return Err("compact must be client, server or off".into()),
        };
        if compact == Compact::Server && provider != Provider::Anthropic {
            return Err("compact server is Anthropic's compaction beta (provider anthropic)".into());
        }
        let window = number(&args.context_window, CONTEXT_WINDOW, 100_000_000).map_err(|e| format!("contextWindow {e}"))?;
        let at = match &args.compact_at {
            None | Some(Value::Null) => Some(COMPACT_AT),
            Some(Value::String(s)) if s.trim().is_empty() => Some(COMPACT_AT),
            Some(Value::String(s)) => s.trim().parse::<f64>().ok(),
            Some(v) => v.as_f64(),
        }
        .filter(|a| *a > 0.0 && *a <= 1.0)
        .ok_or("compactAt must be a number above 0, at most 1")?;
        let dynamic = match opt_json(&args.dynamic_mcp, "dynamicMcp")? {
            None | Some(Value::Bool(false)) => vec![],
            Some(Value::Bool(true)) => vec!["url".into()],
            Some(Value::Array(l)) if l.iter().all(|a| a.as_str().is_some_and(|a| !a.trim().is_empty())) => l.iter().filter_map(|a| a.as_str().map(|a| a.trim().to_string())).collect(),
            Some(_) => return Err("dynamicMcp must be true, false or a list of \"url\", URL prefixes and hook NSIDs".into()),
        };
        Ok(Self {
            provider,
            stream,
            prompt: args.prompt.clone().unwrap_or_default(),
            compact,
            threshold: (window as f64 * at) as u64,
            compact_prompt: args.compact_prompt.clone().filter(|p| !p.trim().is_empty()),
            model: model.unwrap_or_else(|| MODEL.into()),
            url: if provider == Provider::OpenAi { format!("{endpoint}/chat/completions") } else { format!("{endpoint}/v1/messages") },
            max_tokens: number(&args.max_tokens, MAX_TOKENS, 128_000).map_err(|e| format!("maxTokens {e}"))?,
            max_steps: number(&args.max_steps, MAX_STEPS, 1000).map_err(|e| format!("maxSteps {e}"))?,
            system: args.system.clone().filter(|s| !s.trim().is_empty()),
            extra,
            dynamic,
        })
    }
}

/// A string option that may also be JSON (a form field "false"): absent when empty.
fn opt_json_or_text(value: &Option<Value>) -> Option<Value> {
    match value {
        None | Some(Value::Null) => None,
        Some(Value::String(s)) if s.trim().is_empty() => None,
        Some(Value::String(s)) => Some(json!(s.trim().to_ascii_lowercase())),
        Some(v) => Some(v.clone()),
    }
}

/// A JSON value given as itself or as its text (form fields are strings); empty is absent.
fn opt_json(value: &Option<Value>, what: &str) -> Result<Option<Value>, String> {
    match value {
        None | Some(Value::Null) => Ok(None),
        Some(Value::String(s)) if s.trim().is_empty() => Ok(None),
        Some(Value::String(s)) => serde_json::from_str(s).map(Some).map_err(|e| format!("{what}: {e}")),
        Some(v) => Ok(Some(v.clone())),
    }
}

fn number(value: &Option<Value>, default: u64, max: u64) -> Result<u64, String> {
    let n = match value {
        None | Some(Value::Null) => return Ok(default),
        Some(Value::String(s)) if s.trim().is_empty() => return Ok(default),
        Some(Value::String(s)) => s.trim().parse::<u64>().ok(),
        Some(v) => v.as_u64(),
    };
    n.filter(|n| (1..=max).contains(n)).ok_or_else(|| format!("must be a whole number from 1 to {max}"))
}

/// The configured servers; none given: one, the in-process hook (an unrouted hook answers null: skipped).
fn servers(args: &Args) -> Result<Vec<Server>, String> {
    let list = match opt_json(&args.mcp, "mcp")? {
        None => vec![json!({ "name": "mcp" })],
        Some(Value::Array(l)) => l,
        Some(_) => return Err("mcp must be a list of { name, url } or { name, hook }".into()),
    };
    let mut out: Vec<Server> = vec![];
    for s in list {
        let server = server_of(s["name"].as_str().unwrap_or_default(), if s["server"].is_object() { &s["server"] } else { &s })?;
        if out.iter().any(|o| o.name == server.name) {
            return Err(format!("mcp server name {} is used twice", server.name));
        }
        out.push(server);
    }
    if out.len() > MAX_SERVERS {
        return Err(format!("mcp: at most {MAX_SERVERS} servers"));
    }
    Ok(out)
}

/// A server from its name and descriptor: { url } (Streamable HTTP over whatever the host fetches that scheme with)
/// or { hook? } (a composition hook, by default com.fedfork.exec.mcp.message).
fn server_of(name: &str, d: &Value) -> Result<Server, String> {
    if name.is_empty() || name.len() > 32 || !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-') {
        return Err(format!("mcp server name {name:?}: 1 to 32 of A-Z a-z 0-9 _ -"));
    }
    if d.get("nsid").is_some() || d.get("input").is_some() {
        return Err(format!("mcp server {name}: a server is {{url}} or {{hook}} (no nsid descriptors: a host transport is a webrtc:// or iroh:// url)"));
    }
    let [url, hook] = ["url", "hook"].map(|k| d[k].as_str().map(String::from));
    if url.is_some() && hook.is_some() {
        return Err(format!("mcp server {name}: one of url or hook"));
    }
    if url.as_deref().is_some_and(|u| !has_scheme(u)) {
        return Err(format!("mcp server {name}: url must be <scheme>://... (https, or a host transport's webrtc:// or iroh://)"));
    }
    let hook = url.is_none().then(|| hook.unwrap_or_else(|| MCP_HOOK.into()));
    Ok(Server { name: name.into(), url, hook, phase: "init".into(), ..Default::default() })
}

/// <scheme>://..., the scheme [A-Za-z][A-Za-z0-9+.-]*
fn has_scheme(url: &str) -> bool {
    url.split_once("://").is_some_and(|(s, _)| s.starts_with(|c: char| c.is_ascii_alphabetic()) && s.chars().all(|c| c.is_ascii_alphanumeric() || "+.-".contains(c)))
}

/// A response header, whatever its case (a host transport may pass the guest's as sent).
fn header<'a>(output: &'a Value, name: &str) -> Option<&'a str> {
    output["headers"].as_object()?.iter().find(|(k, _)| k.eq_ignore_ascii_case(name))?.1.as_str()
}

fn fail(server: &mut Server, error: String) {
    server.phase = "error".into();
    server.error = Some(error);
}

fn request(id: &str, method: &str, params: Value) -> Value {
    json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params })
}

fn initialize(i: usize, server: &Server) -> Call {
    let id = format!("init.{i}");
    let params = json!({ "protocolVersion": PROTOCOL, "capabilities": {}, "clientInfo": { "name": "atlibs-asyncai", "version": env!("CARGO_PKG_VERSION") } });
    mcp_call(&id, server, request(&id, "initialize", params))
}

fn list(i: usize, page: u64, server: &Server, cursor: Option<&str>) -> Call {
    let id = format!("list.{i}.{page}");
    let params = cursor.map_or_else(|| json!({}), |c| json!({ "cursor": c }));
    mcp_call(&id, server, request(&id, "tools/list", params))
}

/// One JSON-RPC message to a server: a POST of it (Streamable HTTP, its deadline past the call's own wait), or the hook call.
fn mcp_call(id: &str, server: &Server, message: Value) -> Call {
    let Some(url) = &server.url else {
        return Call::new(id, server.hook.as_deref().unwrap_or(MCP_HOOK)).input(message);
    };
    let mut headers = json!({ "content-type": "application/json", "accept": "application/json, text/event-stream" });
    if let Some(v) = &server.version {
        headers["mcp-protocol-version"] = json!(v);
    }
    if let Some(s) = &server.session {
        headers["mcp-session-id"] = json!(s);
    }
    let args = &message["params"]["arguments"];
    let wait = ["waitMs", "timeoutMs"].iter().filter_map(|k| args[k].as_u64()).max().unwrap_or(0);
    Call::new(id, HTTP).input(json!({ "method": "POST", "url": url, "headers": headers, "body": message.to_string() })).timeout_ms(MCP_TIMEOUT + wait.min(MCP_WAIT_MAX))
}

fn clip(text: &str, max: usize) -> String {
    match text.char_indices().nth(max) {
        Some((at, _)) => format!("{}...", &text[..at]),
        None => text.into(),
    }
}

fn http_error(output: &Value) -> Option<String> {
    let status = output["status"].as_u64().unwrap_or(0);
    (!(200..300).contains(&status)).then(|| format!("HTTP {status}: {}", clip(output["body"].as_str().unwrap_or_default().trim(), 300)))
}

/// The messages in an event stream's `data:` fields.
fn sse(body: &str) -> Vec<Value> {
    body.replace("\r\n", "\n")
        .split("\n\n")
        .filter_map(|event| {
            let data: Vec<&str> = event.lines().filter_map(|l| l.strip_prefix("data:")).map(|d| d.strip_prefix(' ').unwrap_or(d)).collect();
            (!data.is_empty()).then(|| serde_json::from_str(&data.join("\n")).ok()).flatten()
        })
        .collect()
}

/// The result of request `id` from a server's answer; Ok(None): a hook nothing routes, or a transport, answered null.
fn reply(server: &Server, id: &str, output: &Value) -> Result<Option<Value>, String> {
    let message = if server.url.is_some() {
        if let Some(e) = http_error(output) {
            return Err(e);
        }
        let body = output["body"].as_str().unwrap_or_default();
        let streamed = header(output, "content-type").is_some_and(|t| t.starts_with("text/event-stream"));
        let messages = if streamed {
            sse(body)
        } else {
            match serde_json::from_str(body).map_err(|e| format!("not a JSON-RPC response: {e}"))? {
                Value::Array(batch) => batch,
                one => vec![one],
            }
        };
        messages.into_iter().find(|m| m["id"] == id && (m.get("result").is_some() || m.get("error").is_some())).ok_or_else(|| format!("no response to {id}"))?
    } else if output.is_null() {
        return Ok(None);
    } else {
        output.clone()
    };
    if let Some(e) = message.get("error") {
        return Err(format!("{} ({})", e["message"].as_str().unwrap_or("error"), e["code"]));
    }
    Ok(Some(message.get("result").cloned().unwrap_or(Value::Null)))
}

/// A notification went through: HTTP 2xx (202 Accepted), or a hook answered (null).
fn notified(server: &Server, output: &Value) -> Result<(), String> {
    match (&server.url, http_error(output)) {
        (Some(_), Some(e)) => Err(e),
        _ => Ok(()),
    }
}

fn settled(server: &Server) -> bool {
    matches!(server.phase.as_str(), "ready" | "skipped" | "error" | "disconnected")
}

/// Every server ready, skipped or failed: name the tools and start inferring. Mid-run (a server mcp_connect
/// started settled): that tool call's result.
fn ready(mut state: State, config: &Config) -> Handle {
    if state.steps > 0 {
        return connected(state, config);
    }
    if !state.sized || !state.servers.iter().all(settled) {
        return Handle::new().state(json!(state));
    }
    for i in 0..state.servers.len() {
        if state.servers[i].phase == "ready" {
            register(&mut state, i);
        }
    }
    if !config.dynamic.is_empty() {
        state.tools.extend(builtins(config));
    }
    infer(state, config)
}

/// Server i's listed tools as <server>__<tool> (Messages API names: ^[a-zA-Z0-9_-]{1,64}$); a longer name or a
/// clash is dropped. Returns the names it added.
fn register(state: &mut State, i: usize) -> Vec<String> {
    let mut added = vec![];
    let server = &mut state.servers[i];
    for tool in &server.listed {
        let name = tool["name"].as_str().unwrap_or_default();
        let namespaced: String = format!("{}__{name}", server.name).chars().map(|c| if c.is_ascii_alphanumeric() || c == '_' || c == '-' { c } else { '_' }).collect();
        if name.is_empty() || namespaced.len() > 64 || state.routes.contains_key(&namespaced) {
            server.dropped.push(name.into());
            continue;
        }
        let mut def = json!({ "name": namespaced, "input_schema": tool.get("inputSchema").cloned().unwrap_or_else(|| json!({ "type": "object" })) });
        if let Some(d) = tool["description"].as_str() {
            def["description"] = json!(d);
        }
        state.tools.push(def);
        state.routes.insert(namespaced.clone(), json!([i, name]));
        added.push(namespaced);
    }
    added
}

/// mcp_connect and mcp_disconnect, their descriptions naming what this run allows.
fn builtins(config: &Config) -> [Value; 2] {
    let allowed = config.dynamic.join(", ");
    let connect = format!(
        "Connect an MCP server for the rest of this run: its tools are offered as <name>__<tool> from your next turn on. \
server is {{\"url\": \"https://...\"}} (Streamable HTTP; also webrtc://... or iroh://..., a VM's MCP server as the requester's \
contract result gives it) or {{\"hook\": \"<nsid>\"}} (an in-process server the composition routes). Allowed here: {allowed} \
(url: any URL the host allows; a URL: URLs under it; an NSID: that hook)."
    );
    [
        json!({ "name": CONNECT, "description": connect, "input_schema": { "type": "object", "properties": {
            "name": { "type": "string", "description": "1 to 32 of A-Z a-z 0-9 _ -, not in use: the prefix of its tools" },
            "server": { "type": "object", "properties": { "url": { "type": "string" }, "hook": { "type": "string" } } }
        }, "required": ["name", "server"] } }),
        json!({ "name": DISCONNECT, "description": "Disconnect an MCP server (one given at the start or connected with mcp_connect): its tools are gone from your next turn on.",
            "input_schema": { "type": "object", "properties": { "name": { "type": "string" } }, "required": ["name"] } }),
    ]
}

/// The URL is the prefix or under it (a path boundary: https://a.example does not allow https://a.example.evil;
/// webrtc:/// allows every webrtc URL).
fn under(url: &str, prefix: &str) -> bool {
    let base = prefix.trim_end_matches('/');
    has_scheme(prefix) && (url == base || url.starts_with(&format!("{base}/")))
}

/// mcp_connect: a new server (initialize goes out with this output), when this run allows its descriptor.
fn connect(state: &mut State, config: &Config, input: &Value) -> Result<usize, String> {
    let name = input["name"].as_str().unwrap_or_default();
    if !input["server"].is_object() {
        return Err("server must be {url} or {hook}".into());
    }
    let mut server = server_of(name, &input["server"])?;
    if state.servers.iter().any(|s| s.name == name && s.phase != "disconnected") {
        return Err(format!("an MCP server named {name} is connected: disconnect it first, or pick another name"));
    }
    if state.servers.len() >= MAX_SERVERS {
        return Err(format!("this run has had {MAX_SERVERS} MCP servers, the most it may"));
    }
    let (what, allowed) = match (&server.url, &server.hook) {
        (Some(u), _) => (u, config.dynamic.iter().any(|a| a == "url" || under(u, a))),
        (_, Some(n)) => (n, config.dynamic.contains(n)),
        _ => unreachable!("server_of gives one"),
    };
    if !allowed {
        return Err(format!("{what} is not allowed here (allowed: {})", config.dynamic.join(", ")));
    }
    server.dynamic = true;
    state.servers.push(server);
    Ok(state.servers.len() - 1)
}

/// mcp_disconnect: the server's tools are gone from the next request on (its tool calls already out still answer).
fn disconnect(state: &mut State, input: &Value) -> Result<String, String> {
    let name = input["name"].as_str().unwrap_or_default();
    let i = state.servers.iter().position(|s| s.name == name && s.phase != "disconnected").ok_or_else(|| format!("no MCP server named {name:?}"))?;
    if !settled(&state.servers[i]) {
        return Err(format!("{name} is still connecting"));
    }
    state.servers[i].phase = "disconnected".into();
    let routes = &state.routes;
    state.tools.retain(|t| routes.get(t["name"].as_str().unwrap_or_default()).is_none_or(|r| r[0] != i));
    state.routes.retain(|_, r| r[0] != i);
    Ok(format!("disconnected {name}: its tools are gone from your next turn on"))
}

/// The servers this turn's mcp_connects started are all ready (their tools, from the next request on, in
/// tool_use order whichever answered first) or failed (no longer in the live set).
fn connected(mut state: State, config: &Config) -> Handle {
    let connects: Vec<(usize, String, usize)> = state.waiting.iter().filter(|w| w.0.starts_with("connect.") && state.results[w.4].is_null()).map(|w| (w.2, w.1.clone(), w.4)).collect();
    if connects.iter().any(|(i, ..)| !settled(&state.servers[*i])) {
        return Handle::new().state(json!(state));
    }
    for (i, use_id, k) in connects {
        let outcome = if state.servers[i].phase == "ready" {
            let tools = register(&mut state, i);
            let s = &state.servers[i];
            let mut out = json!({ "connected": s.name, "protocolVersion": s.version, "tools": tools });
            if !s.dropped.is_empty() {
                out["dropped"] = json!(s.dropped);
            }
            (out.to_string(), false)
        } else {
            let s = &mut state.servers[i];
            let why = s.error.clone().unwrap_or_else(|| "nothing answered initialize (no route or transport)".into());
            s.phase = "disconnected".into();
            (format!("not connected: {why}"), true)
        };
        state.results[k] = tool_result(config, &use_id, outcome);
    }
    answered(state, config)
}

fn infer(mut state: State, config: &Config) -> Handle {
    // the next request past the threshold: summarize first (not mid-pause_turn: the last message is the user's)
    let due = config.compact == Compact::Client && !state.fresh && state.outbox.last().is_some_and(|m| m["role"] == "user" || m["role"] == "tool");
    if due && estimate(&state) > config.threshold {
        return compact(state, config);
    }
    state.fresh = false;
    state.steps += 1;
    state.text.clear();
    let last = state.steps >= config.max_steps;
    let mut body = body(&state, config);
    if !state.tools.is_empty() && last && state.steps > 1 {
        if config.provider == Provider::OpenAi {
            body["tool_choice"] = json!("none");
            state.outbox.push(json!({ "role": "user", "content": TOOL_BUDGET_EXHAUSTED }));
        } else {
            body["tool_choice"] = json!({ "type": "none" });
            // the last user message is this turn's tool results, not sent yet
            if let Some(m) = state.outbox.last_mut() {
                add_text(m, TOOL_BUDGET_EXHAUSTED);
            }
        }
    }
    let call = post(format!("infer.{}", state.steps), config, body);
    let handle = append(&mut state, config, Handle::new());
    handle.call(call).state(json!(state))
}

/// A request body naming the held transcript, the extra fields over it. OpenAI chat completions: the held
/// system message then the transcript, the tools as functions (an MCP inputSchema is its parameters).
fn body(state: &State, config: &Config) -> Value {
    let mut body = json!({ "model": config.model, "max_tokens": config.max_tokens });
    if config.provider == Provider::OpenAi {
        body["messages"] = json!({ "$held": [SYSTEM, TRANSCRIPT] });
        if !state.tools.is_empty() {
            let functions: Vec<Value> = state
                .tools
                .iter()
                .map(|t| {
                    let mut f = json!({ "name": t["name"], "parameters": t["input_schema"] });
                    if let Some(d) = t.get("description") {
                        f["description"] = d.clone();
                    }
                    json!({ "type": "function", "function": f })
                })
                .collect();
            body["tools"] = json!(functions);
        }
        if config.stream {
            body["stream"] = json!(true);
            body["stream_options"] = json!({ "include_usage": true });
        }
        for (k, v) in &config.extra {
            body[k] = v.clone();
        }
        return body;
    }
    if let Some(s) = &config.system {
        body["system"] = json!(s);
    }
    if !state.tools.is_empty() {
        body["tools"] = json!(state.tools);
    }
    body["messages"] = json!({ "$held": TRANSCRIPT });
    if config.compact == Compact::Server {
        let mut edit = json!({ "type": "compact_20260112", "trigger": { "type": "input_tokens", "value": config.threshold.max(SERVER_COMPACT_MIN) } });
        if let Some(p) = &config.compact_prompt {
            edit["instructions"] = json!(p);
        }
        body["context_management"] = json!({ "edits": [edit] });
    }
    for (k, v) in &config.extra {
        body[k] = v.clone();
    }
    body
}

/// The body as JSON: the host splices the transcript in (after this output's append).
fn post(id: String, config: &Config, body: Value) -> Call {
    if config.provider == Provider::OpenAi {
        let accept = if config.stream { "text/event-stream" } else { "application/json" };
        let headers = json!({ "content-type": "application/json", "accept": accept });
        return Call::new(id, HTTP).input(json!({ "method": "POST", "url": config.url, "headers": headers, "body": body }));
    }
    let mut headers = json!({ "content-type": "application/json", "anthropic-version": ANTHROPIC_VERSION });
    if config.compact == Compact::Server {
        headers["anthropic-beta"] = json!(SERVER_COMPACT_BETA);
    }
    Call::new(id, HTTP).input(json!({ "method": "POST", "url": config.url, "headers": headers, "body": body }))
}

/// A text block after a message's content (a string content becomes a text block first).
fn add_text(message: &mut Value, text: &str) {
    if let Some(s) = message["content"].as_str() {
        message["content"] = json!([{ "type": "text", "text": s }]);
    }
    if let Some(content) = message["content"].as_array_mut() {
        content.push(json!({ "type": "text", "text": text }));
    }
}

/// The next request's tokens: the last one's input and output, and about 4 bytes a token for what is new.
fn estimate(state: &State) -> u64 {
    state.context + serde_json::to_string(&state.outbox).map_or(0, |s| s.len() as u64 / 4)
}

/// Summarize the transcript: everything held and the outbox, the instruction in the last user message. It keeps the
/// messages from the last tool_use turn on (the tool_use with its tool_result), which the module still has; with
/// none in the outbox (a resumed conversation at the start), only the summary and the prompt go on.
fn compact(mut state: State, config: &Config) -> Handle {
    let uses = |m: &Value| m["role"] == "assistant" && (m["content"].as_array().is_some_and(|c| c.iter().any(|b| b["type"] == "tool_use")) || m["tool_calls"].as_array().is_some_and(|c| !c.is_empty()));
    let from = state.outbox.iter().rposition(uses).unwrap_or(state.outbox.len());
    state.kept = Some(state.outbox[from..].to_vec());
    let instruction = config.compact_prompt.as_deref().unwrap_or(COMPACT_PROMPT);
    match state.outbox.last_mut() {
        // after tool messages, a user message of its own
        Some(m) if config.provider == Provider::OpenAi && m["role"] == "tool" => state.outbox.push(json!({ "role": "user", "content": instruction })),
        Some(m) if config.provider == Provider::OpenAi => m["content"] = json!(format!("{}\n\n{instruction}", m["content"].as_str().unwrap_or_default())),
        Some(m) => add_text(m, instruction),
        None => {}
    }
    state.compactions += 1;
    let handle = append(&mut state, config, Handle::new());
    let call = summary_call(&state, config);
    handle.call(call).state(json!(state))
}

fn summary_call(state: &State, config: &Config) -> Call {
    post(format!("compact.{}", state.compactions), config, body(state, config))
}

fn add_usage(state: &mut State, response: &Value) {
    for (k, v) in response["usage"].as_object().into_iter().flatten() {
        if let Some(n) = v.as_u64() {
            let sum = state.usage.get(k).and_then(Value::as_u64).unwrap_or(0) + n;
            state.usage.insert(k.clone(), json!(sum));
        }
    }
}

/// A 429/5xx to retry (after a timer), or the response's JSON.
enum Answer {
    Retry(Handle),
    Failed(String),
    Body(Value),
}

fn answer(state: &mut State, output: Result<Value, String>, timer: &str) -> Answer {
    let output = match output {
        Ok(o) => o,
        Err(e) => return Answer::Failed(format!(": {e}")),
    };
    let status = output["status"].as_u64().unwrap_or(0);
    if matches!(status, 429 | 500 | 502 | 503 | 504 | 529) && state.retries < RETRIES {
        let after = header(&output, "retry-after").and_then(|s| s.trim().parse::<u64>().ok()).map(|s| s.min(120) * 1000);
        let ms = after.unwrap_or(2000 * 4u64.pow(state.retries as u32));
        state.retries += 1;
        return Answer::Retry(Handle::new().timer(timer, ms));
    }
    if let Some(e) = http_error(&output) {
        return Answer::Failed(format!(" {e}"));
    }
    state.retries = 0;
    let body = output["body"].as_str().unwrap_or_default();
    if header(&output, "content-type").is_some_and(|t| t.starts_with("text/event-stream")) {
        return collected(body).map_or_else(Answer::Failed, Answer::Body);
    }
    match serde_json::from_str(body) {
        Ok(r) => Answer::Body(r),
        Err(e) => Answer::Failed(format!(" answered no JSON: {e}")),
    }
}

/// An OpenAI chat.completion.chunk stream, collected into the chat.completion it adds up to: the content
/// appended, each tool call's name and arguments by index, the last finish_reason and usage.
fn collected(body: &str) -> Result<Value, String> {
    let (mut content, mut finish, mut usage, mut model) = (String::new(), Value::Null, Value::Null, Value::Null);
    let mut calls: Vec<Value> = vec![];
    for chunk in sse(body) {
        if let Some(e) = chunk.get("error") {
            return Err(format!(" stream error: {}", e["message"].as_str().unwrap_or("unknown")));
        }
        let choice = &chunk["choices"][0];
        content.push_str(choice["delta"]["content"].as_str().unwrap_or_default());
        for part in choice["delta"]["tool_calls"].as_array().into_iter().flatten() {
            let i = part["index"].as_u64().unwrap_or(0) as usize;
            while calls.len() <= i {
                calls.push(json!({ "id": "", "type": "function", "function": { "name": "", "arguments": "" } }));
            }
            let call = &mut calls[i];
            if let Some(id) = part["id"].as_str() {
                call["id"] = json!(id);
            }
            for key in ["name", "arguments"] {
                let piece = part["function"][key].as_str().unwrap_or_default();
                call["function"][key] = json!(format!("{}{piece}", call["function"][key].as_str().unwrap_or_default()));
            }
        }
        for (slot, value) in [(&mut finish, &choice["finish_reason"]), (&mut usage, &chunk["usage"]), (&mut model, &chunk["model"])] {
            if !value.is_null() {
                *slot = value.clone();
            }
        }
    }
    let mut message = json!({ "role": "assistant", "content": content });
    if !calls.is_empty() {
        message["tool_calls"] = json!(calls);
    }
    Ok(json!({ "model": model, "choices": [{ "index": 0, "message": message, "finish_reason": finish }], "usage": usage }))
}

/// The summary in: the held transcript becomes it, then the kept messages (their thinking dropped: it was
/// bound to the history it no longer follows), then the next request.
fn compacted(mut state: State, config: &Config, output: Result<Value, String>) -> Handle {
    let response = match answer(&mut state, output, "recompact") {
        Answer::Retry(h) => return h.state(json!(state)),
        Answer::Failed(e) => return finish(state, config, "error", Some(format!("compaction{e}"))),
        Answer::Body(r) => r,
    };
    add_usage(&mut state, &response);
    let text = turn(&response, config.provider).text.join("\n");
    let summary = text.split_once("<summary>").map_or(text.as_str(), |(_, s)| s.split("</summary>").next().unwrap_or(s)).trim();
    if summary.is_empty() {
        return finish(state, config, "error", Some(format!("compaction: no summary (stop_reason {})", response["stop_reason"])));
    }
    let kept: Vec<Value> = state
        .kept
        .take()
        .unwrap_or_default()
        .into_iter()
        .filter_map(|mut m| {
            let assistant = m["role"] == "assistant";
            if let Some(o) = m.as_object_mut().filter(|_| assistant) {
                o.remove("reasoning_content");
            }
            if let Some(c) = m["content"].as_array_mut().filter(|_| assistant) {
                c.retain(|b| b["type"] != "thinking" && b["type"] != "redacted_thinking");
                if c.is_empty() {
                    return None;
                }
            }
            Some(m)
        })
        .collect();
    let mut first = format!("{SUMMARY_PREFIX}\n\n{summary}");
    // the servers the model connected are still connected (their tools still offered), which the summary may not say
    let live: Vec<&str> = state.servers.iter().filter(|s| s.dynamic && s.phase == "ready").map(|s| s.name.as_str()).collect();
    if !live.is_empty() {
        first.push_str(&format!("\n\nMCP servers connected with {CONNECT}, still connected (their tools are <name>__<tool>): {}.", live.join(", ")));
    }
    if !kept.iter().any(|m| m["content"] == json!(config.prompt)) {
        first.push_str(&format!("\n\nThe user's latest request: {}", config.prompt));
    }
    state.base = Some(vec![json!({ "role": "user", "content": first })]);
    state.outbox = kept;
    state.context = 0;
    state.fresh = true;
    infer(state, config)
}

/// The outbox appended to the held transcript (after a compaction's new start), ahead of whatever else this output calls.
fn append(state: &mut State, config: &Config, mut handle: Handle) -> Handle {
    if !state.started && config.provider == Provider::OpenAi {
        let system: Vec<Value> = config.system.iter().map(|s| json!({ "role": "system", "content": s })).collect();
        state.next += 1;
        handle = handle.call(Call::new(format!("held.{}", state.next), PUT).input(json!({ "name": SYSTEM, "value": system })));
    }
    state.started = true;
    if let Some(base) = state.base.take() {
        state.next += 1;
        handle = handle.call(Call::new(format!("held.{}", state.next), PUT).input(json!({ "name": TRANSCRIPT, "value": base })));
    }
    if state.outbox.is_empty() {
        return handle;
    }
    state.next += 1;
    handle.call(Call::new(format!("held.{}", state.next), APPEND).input(json!({ "name": TRANSCRIPT, "items": std::mem::take(&mut state.outbox) })))
}

/// One inference answer, either API's: the assistant message to keep (in that API's form), its text, its tool
/// calls (id, name, input), the stop reason in Messages API terms, and the transcript's size in tokens.
struct Turn {
    message: Value,
    text: Vec<String>,
    uses: Vec<(String, String, Value)>,
    stop: String,
    context: u64,
}

fn turn(response: &Value, provider: Provider) -> Turn {
    let u = &response["usage"];
    if provider == Provider::OpenAi {
        let choice = &response["choices"][0];
        let m = &choice["message"];
        let calls = m["tool_calls"].as_array().cloned().unwrap_or_default();
        let text: Vec<String> = m["content"].as_str().filter(|t| !t.is_empty()).map(String::from).into_iter().collect();
        let mut message = json!({ "role": "assistant", "content": m["content"].as_str().unwrap_or_default() });
        if !calls.is_empty() {
            message["tool_calls"] = json!(calls);
        }
        let uses = calls
            .iter()
            .map(|c| {
                let args = c["function"]["arguments"].as_str().and_then(|a| serde_json::from_str(a).ok()).filter(Value::is_object).unwrap_or_else(|| json!({}));
                (c["id"].as_str().unwrap_or_default().to_string(), c["function"]["name"].as_str().unwrap_or_default().to_string(), args)
            })
            .collect::<Vec<_>>();
        let stop = match choice["finish_reason"].as_str().unwrap_or("stop") {
            _ if !uses.is_empty() => "tool_use",
            "stop" => "end_turn",
            "length" => "max_tokens",
            "content_filter" => "refusal",
            other => other,
        };
        let context = u["total_tokens"].as_u64().unwrap_or_else(|| u["prompt_tokens"].as_u64().unwrap_or(0) + u["completion_tokens"].as_u64().unwrap_or(0));
        return Turn { message, text, uses, stop: stop.into(), context };
    }
    let content = response["content"].as_array().cloned().unwrap_or_default();
    Turn {
        text: content.iter().filter(|b| b["type"] == "text").filter_map(|b| b["text"].as_str().map(String::from)).collect(),
        uses: content.iter().filter(|b| b["type"] == "tool_use").map(|b| (b["id"].as_str().unwrap_or_default().into(), b["name"].as_str().unwrap_or_default().into(), b["input"].clone())).collect(),
        message: json!({ "role": "assistant", "content": content }),
        stop: response["stop_reason"].as_str().unwrap_or("end_turn").into(),
        context: ["input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens", "output_tokens"].iter().filter_map(|k| u[k].as_u64()).sum(),
    }
}

fn inferred(mut state: State, config: &Config, output: Result<Value, String>) -> Handle {
    let mut response = match answer(&mut state, output, "retry") {
        Answer::Retry(h) => {
            state.steps -= 1;
            return h.state(json!(state));
        }
        Answer::Failed(e) => return finish(state, config, "error", Some(format!("inference{e}"))),
        Answer::Body(r) => r,
    };
    add_usage(&mut state, &response);
    // server-side compaction: the API ignores everything before its compaction block, so the held transcript starts there
    if let Some(content) = response["content"].as_array_mut().filter(|_| config.provider == Provider::Anthropic) {
        if let Some(at) = content.iter().position(|b| b["type"] == "compaction" && !b["content"].is_null()) {
            content.drain(..at);
            state.base = Some(vec![]);
            state.compactions += 1;
        }
    }
    let Turn { message, text, uses, stop, context } = turn(&response, config.provider);
    state.context = context;
    state.text = text;
    state.outbox.push(message);
    match stop.as_str() {
        "pause_turn" if state.steps < config.max_steps => infer(state, config),
        "tool_use" => {
            if state.steps >= config.max_steps {
                // keep the transcript valid for the next run: every tool_use gets its result
                let results: Vec<Value> = uses.iter().map(|(id, ..)| tool_result(config, id, ("not run: the step cap was reached".into(), true))).collect();
                push_results(&mut state, config, results);
                return finish(state, config, "max_steps", None);
            }
            if uses.is_empty() {
                return finish(state, config, &stop, None);
            }
            let mut handle = Handle::new();
            state.waiting.clear();
            state.results = vec![Value::Null; uses.len()];
            for (k, (use_id, name, input)) in uses.iter().enumerate() {
                if !config.dynamic.is_empty() && name == CONNECT {
                    match connect(&mut state, config, input) {
                        Ok(i) => {
                            handle = handle.call(initialize(i, &state.servers[i]));
                            state.waiting.push((format!("connect.{i}"), use_id.clone(), i, name.clone(), k));
                        }
                        Err(e) => state.results[k] = tool_result(config, use_id, (format!("not connected: {e}"), true)),
                    }
                    continue;
                }
                if !config.dynamic.is_empty() && name == DISCONNECT {
                    state.results[k] = tool_result(config, use_id, disconnect(&mut state, input).map_or_else(|e| (e, true), |t| (t, false)));
                    continue;
                }
                state.next += 1;
                let call_id = format!("tool.{}", state.next);
                match state.routes.get(name).and_then(|r| Some((r[0].as_u64()? as usize, r[1].as_str()?.to_string()))) {
                    Some((i, tool)) => {
                        let arguments = if input.is_object() { input.clone() } else { json!({}) };
                        let message = request(&call_id, "tools/call", json!({ "name": tool, "arguments": arguments }));
                        handle = handle.call(mcp_call(&call_id, &state.servers[i], message));
                        state.waiting.push((call_id, use_id.clone(), i, tool, k));
                    }
                    None => state.results[k] = tool_result(config, use_id, (format!("unknown tool {name}"), true)),
                }
            }
            let pending = state.results.iter().any(Value::is_null);
            if !pending {
                return answered(state, config);
            }
            handle.state(json!(state))
        }
        _ => finish(state, config, &stop, None),
    }
}

/// A tool's result: a Messages API tool_result block, or an OpenAI tool message (no error flag: the text says so).
fn tool_result(config: &Config, use_id: &str, (text, is_error): (String, bool)) -> Value {
    let text = if text.is_empty() { "(no output)".to_string() } else { text };
    if config.provider == Provider::OpenAi {
        return json!({ "role": "tool", "tool_call_id": use_id, "content": if is_error { format!("Error: {text}") } else { text } });
    }
    let mut block = json!({ "type": "tool_result", "tool_use_id": use_id, "content": text });
    if is_error {
        block["is_error"] = json!(true);
    }
    block
}

/// A turn's tool results into the outbox: one user message of tool_result blocks, or a tool message each.
fn push_results(state: &mut State, config: &Config, results: Vec<Value>) {
    if results.is_empty() {
        return;
    }
    if config.provider == Provider::OpenAi {
        state.outbox.extend(results);
    } else {
        state.outbox.push(json!({ "role": "user", "content": results }));
    }
}

/// A tools/call result as tool_result text: structuredContent (the exact JSON) when present, else
/// the content blocks, text as is and anything else as its JSON (asyncai format_result).
fn tool_text(result: &Value) -> (String, bool) {
    let is_error = result["isError"] == true;
    if let Some(s) = result.get("structuredContent").filter(|s| !s.is_null()) {
        return (s.to_string(), is_error);
    }
    let parts: Vec<String> = result["content"].as_array().into_iter().flatten().map(|b| if b["type"] == "text" { b["text"].as_str().unwrap_or_default().to_string() } else { b.to_string() }).collect();
    (parts.join("\n"), is_error)
}

/// One tool result in: when every one of this turn's is, they go back as one user message.
fn answered(mut state: State, config: &Config) -> Handle {
    if state.results.iter().any(Value::is_null) {
        return Handle::new().state(json!(state));
    }
    let results = std::mem::take(&mut state.results);
    push_results(&mut state, config, results);
    state.waiting.clear();
    infer(state, config)
}

fn finish(mut state: State, config: &Config, stop: &str, error: Option<String>) -> Handle {
    // the text of the last message, when the model's
    let text = if state.outbox.last().is_some_and(|m| m["role"] != "assistant") { vec![] } else { state.text.clone() };
    // the live set: the servers given at the start and those connected since, less the disconnected
    let servers: Vec<Value> = state
        .servers
        .iter()
        .filter(|s| s.phase != "disconnected")
        .map(|s| {
            let mut out = json!({ "name": s.name, "phase": s.phase, "tools": s.listed.iter().filter_map(|t| t["name"].as_str()).collect::<Vec<_>>() });
            for (k, v) in [("url", s.url.as_ref()), ("hook", s.hook.as_ref()), ("protocolVersion", s.version.as_ref()), ("error", s.error.as_ref())] {
                if let Some(v) = v {
                    out[k] = json!(v);
                }
            }
            if s.dynamic {
                out["dynamic"] = json!(true);
            }
            if !s.dropped.is_empty() {
                out["dropped"] = json!(s.dropped);
            }
            out
        })
        .collect();
    let mut result = json!({
        "text": text.join("\n"),
        "stop": stop,
        "steps": state.steps,
        "usage": state.usage,
        "compactions": state.compactions,
        "servers": servers,
        "tools": state.tools.iter().filter_map(|t| t["name"].as_str()).collect::<Vec<_>>(),
        "messages": { "$held": TRANSCRIPT },
    });
    if let Some(e) = error {
        result["error"] = json!(e);
    }
    // the host fills in the transcript when the run ends, after this output's append
    append(&mut state, config, Handle::new()).result(result).state(Value::Null).done()
}