iron-shell 14.0.2

//! Agentic AI Callable API for IronShell
//!
//! This module provides a structured JSON-based API for AI agents to interact
//! with IronShell without needing to generate multi-line code blocks.
//!
//! Features:
//! - Single-call builtin execution
//! - Pipeline construction and execution
//! - Language ontology/schema export
//! - Native support for OpenAI, Claude, and Gemini function calling formats
//!
//! # AI Agent Integration
//!
//! Instead of generating brittle IronShell code:
//! ```ish
//! let files = ls(".")
//! files | where(fn(f) => f.size > 1000) | map(fn(f) => f.name)
//! ```
//!
//! AI agents can use structured JSON:
//! ```json
//! {
//!   "action": "call",
//!   "builtin": "ls",
//!   "args": { "path": "." }
//! }
//! ```

use anyhow::{anyhow, Result};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value as JsonValue};
use std::collections::{HashMap, HashSet};
use std::sync::{Mutex, OnceLock};

use crate::ast::{Expr, Stmt};
use crate::builtins::BUILTIN_LOOKUP;
use crate::eval::{eval_expr, eval_program};
use crate::marketplace::{RegistryClient, SearchQuery, SortBy};
use crate::modules::all_modules;
use crate::parser::parse_program;
use crate::value::Value;

// ============================================================================
// Core API Types
// ============================================================================

/// Agent API request - the main entry point for AI agents
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "action", rename_all = "snake_case")]
pub enum AgentRequest {
    /// Execute a single builtin function
    Call {
        builtin: String,
        #[serde(default)]
        args: JsonValue,
    },
    /// Execute a pipeline of operations
    Pipeline {
        steps: Vec<PipelineStep>,
        #[serde(default)]
        input: Option<JsonValue>,
    },
    /// Execute raw IronShell code (escape hatch)
    Eval { code: String },
    /// Get information about a builtin
    Describe { builtin: String },
    /// List all available builtins
    ListBuiltins {
        #[serde(default)]
        category: Option<String>,
    },
    /// Get the language ontology/schema, optionally narrowed to one module and
    /// to the builtins whose name or description contains `q`
    Schema {
        #[serde(default)]
        format: SchemaFormat,
        #[serde(default)]
        module: Option<String>,
        #[serde(default)]
        q: Option<String>,
    },
    /// Get type information
    TypeInfo {
        #[serde(default)]
        type_name: Option<String>,
    },
}

/// A single step in a pipeline
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PipelineStep {
    /// Builtin function to call
    pub builtin: String,
    /// Arguments (can be positional array or named object)
    #[serde(default)]
    pub args: JsonValue,
    /// Optional field selector (e.g., "name" to extract .name from results)
    #[serde(default)]
    pub select: Option<String>,
    /// Optional predicate for filtering (simplified expression)
    #[serde(default)]
    pub predicate: Option<String>,
}

/// Schema output format
#[derive(Debug, Clone, Serialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum SchemaFormat {
    /// JSON Schema format
    #[default]
    JsonSchema,
    /// OpenAI function calling format (GPT-4, GPT-4o, o1, o3)
    OpenAI,
    /// Anthropic Claude tool use format (Claude 3.5 Sonnet, Claude 3 Opus)
    Claude,
    /// Google Gemini function declaration format (Gemini 1.5, Gemini 2.0)
    Gemini,
    /// Meta Llama format (Llama 3.1, 3.2, 3.3 via Ollama/vLLM/Together)
    Llama,
    /// Mistral AI function calling format (Mistral Large, Codestral)
    Mistral,
    /// Cohere Command format (Command R+, Command R)
    Cohere,
    /// xAI Grok format (Grok-2)
    Grok,
    /// DeepSeek format (DeepSeek-V3, DeepSeek-R1)
    DeepSeek,
    /// Amazon Bedrock format (Claude, Llama, Titan via Bedrock)
    Bedrock,
    /// Azure OpenAI format (Azure-hosted OpenAI models)
    AzureOpenAI,
    /// Alibaba Qwen format (Qwen 2.5)
    Qwen,
    /// Ollama local format (any model via Ollama)
    Ollama,
    /// vLLM format (high-performance inference)
    VLLM,
    /// HuggingFace TGI format (Text Generation Inference)
    HuggingFace,
    /// OpenRouter format (unified API for multiple providers)
    OpenRouter,
    /// Moonshot AI Kimi format (Kimi, Moonshot)
    Kimi,
    /// 01.AI Yi format (Yi-Large, Yi-Lightning)
    Yi,
    /// Zhipu AI GLM format (ChatGLM, GLM-4)
    GLM,
    /// Reka AI format (Reka Core, Reka Flash)
    Reka,
    /// AI21 Labs format (Jamba, Jurassic)
    AI21,
    /// Perplexity AI format (sonar models)
    Perplexity,
    /// Together AI format (hosted inference)
    Together,
    /// Groq format (fast inference - distinct from xAI Grok)
    Groq,
    /// Fireworks AI format (fast inference)
    Fireworks,
    /// Compact ontology format
    Ontology,
    /// JSON-LD semantic web format
    JsonLD,
    /// OWL/RDF Turtle format
    OwlTurtle,
    /// SHACL validation shapes
    SHACL,
}

/// A schema format is read from the name `ish agent schema -f` takes (`openai`,
/// `claude`, ...). The serde names of the variants (`open_a_i`, `json_schema`)
/// are accepted too, as `parse_schema_format` accepts them.
impl<'de> Deserialize<'de> for SchemaFormat {
    fn deserialize<D: serde::Deserializer<'de>>(
        deserializer: D,
    ) -> std::result::Result<Self, D::Error> {
        let name = String::deserialize(deserializer)?;
        parse_schema_format(&name).ok_or_else(|| {
            serde::de::Error::custom(format!(
                "unknown format '{name}'; expected one of: {SCHEMA_FORMATS_HINT}"
            ))
        })
    }
}

/// Agent API response
///
/// On the wire a failure's `error` is an object (`code`, `message`, `hint`,
/// `retryable`, and the fields of a structured error), built from the fields
/// below by `AgentResponseWire`. Rust callers read the message as a string.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(into = "AgentResponseWire", from = "AgentResponseWire")]
pub struct AgentResponse {
    /// Whether the operation succeeded
    pub success: bool,
    /// The result value (if success)
    pub result: Option<JsonValue>,
    /// Error message (if failure)
    pub error: Option<String>,
    /// Machine-readable error code such as `E_BAD_ARG` (if failure with a known code)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub code: Option<String>,
    /// What to do about the failure, from the structured error (if it has one)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub hint: Option<String>,
    /// Type of the result
    #[serde(skip_serializing_if = "Option::is_none")]
    pub result_type: Option<String>,
    /// Metadata about the operation
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metadata: Option<JsonValue>,
}

/// The wire form of `AgentResponse`. The field order is the order the response
/// has always been written in.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct AgentResponseWire {
    success: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    result: Option<JsonValue>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    error: Option<JsonValue>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    code: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    hint: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    result_type: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    metadata: Option<JsonValue>,
}

impl From<AgentResponse> for AgentResponseWire {
    fn from(r: AgentResponse) -> Self {
        // A failure always has an error object, even one whose message is missing.
        let error = if r.success && r.error.is_none() {
            None
        } else {
            let text = r.error.as_deref().unwrap_or("the request failed");
            Some(error_object(text, r.code.as_deref(), r.hint.as_deref()))
        };
        AgentResponseWire {
            success: r.success,
            result: r.result,
            error,
            code: r.code,
            hint: r.hint,
            result_type: r.result_type,
            metadata: r.metadata,
        }
    }
}

impl From<AgentResponseWire> for AgentResponse {
    fn from(w: AgentResponseWire) -> Self {
        let (error, code, hint) = match &w.error {
            None => (None, w.code.clone(), w.hint.clone()),
            Some(JsonValue::String(text)) => (Some(text.clone()), w.code.clone(), w.hint.clone()),
            Some(object) => {
                let text = object
                    .get("message")
                    .and_then(JsonValue::as_str)
                    .map(str::to_string);
                let code = object
                    .get("code")
                    .and_then(JsonValue::as_str)
                    .map(str::to_string)
                    .or_else(|| w.code.clone());
                let hint = object
                    .get("hint")
                    .and_then(JsonValue::as_str)
                    .map(str::to_string)
                    .or_else(|| w.hint.clone());
                (text, code, hint)
            }
        };
        AgentResponse {
            success: w.success,
            result: w.result,
            error,
            code,
            hint,
            result_type: w.result_type,
            metadata: w.metadata,
        }
    }
}

/// The hint for a failure that has none of its own.
const DEFAULT_ERROR_HINT: &str = "read error.message; error.code names the class of failure, and error.retryable says whether a corrected call can succeed";

/// Whether a failure with this code can succeed once the call is corrected.
/// The same classes `ErrorCode::retryable` gives, by wire code, except the
/// codes `same_request_fails_again` names.
fn code_retryable(code: &str) -> bool {
    matches!(
        code,
        "E_UNKNOWN_FIELD" | "E_NOT_FOUND" | "E_BAD_STATE" | "E_NEEDS_APPROVAL" | "E_NETWORK"
    )
}

/// Codes for which the same request fails the same way when it is sent again:
/// the arguments, the source or the name are what is wrong, and a repeat of the
/// request is never retryable. The structured error's own `retryable` does not
/// change that.
fn same_request_fails_again(code: &str) -> bool {
    matches!(
        code,
        "E_PARSE" | "E_BAD_ARG" | "E_UNKNOWN_BUILTIN" | "E_UNKNOWN_NAME"
    )
}

/// The `error` object of a failed response: `code`, `message`, `hint` and
/// `retryable`. A structured error (`{"error":{...}}`, as the evaluator writes
/// it) contributes its own fields, and any text before it, such as the name of
/// the pipeline step that failed, stays in front of its message.
fn error_object(text: &str, code: Option<&str>, hint: Option<&str>) -> JsonValue {
    let mut fields = match structured_error_body(text) {
        Some((prefix, mut body)) => {
            if !prefix.is_empty() {
                let inner = body
                    .get("message")
                    .and_then(JsonValue::as_str)
                    .unwrap_or_default()
                    .to_string();
                body.insert("message".to_string(), json!(format!("{prefix}: {inner}")));
            }
            body
        }
        None => serde_json::Map::new(),
    };
    if !fields.contains_key("message") {
        fields.insert("message".to_string(), json!(text));
    }
    let code = fields
        .get("code")
        .and_then(JsonValue::as_str)
        .map(str::to_string)
        .or_else(|| code.map(str::to_string))
        .unwrap_or_else(|| "E_UNKNOWN".to_string());
    let hint = fields
        .get("hint")
        .and_then(JsonValue::as_str)
        .map(str::to_string)
        .or_else(|| hint.map(str::to_string))
        .unwrap_or_else(|| DEFAULT_ERROR_HINT.to_string());
    let retryable = !same_request_fails_again(&code)
        && fields
            .get("retryable")
            .and_then(JsonValue::as_bool)
            .unwrap_or_else(|| code_retryable(&code));
    fields.insert("code".to_string(), json!(code));
    fields.insert("hint".to_string(), json!(hint));
    fields.insert("retryable".to_string(), json!(retryable));
    JsonValue::Object(fields)
}

/// The structured error inside `text`, whole or after a prefix, as its
/// `error` object and the prefix text before it.
fn structured_error_body(text: &str) -> Option<(String, serde_json::Map<String, JsonValue>)> {
    text.match_indices('{').find_map(|(start, _)| {
        let value = serde_json::Deserializer::from_str(&text[start..])
            .into_iter::<JsonValue>()
            .next()?
            .ok()?;
        let body = value.get("error")?.as_object()?;
        body.get("code").and_then(JsonValue::as_str)?;
        let prefix = text[..start]
            .trim()
            .trim_end_matches(':')
            .trim()
            .to_string();
        Some((prefix, body.clone()))
    })
}

// ============================================================================
// Language Ontology Types
// ============================================================================

/// Complete language ontology for AI agents
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LanguageOntology {
    /// Ontology schema version (semver)
    pub ontology_version: String,
    /// Language metadata
    pub language: LanguageInfo,
    /// Type system
    pub types: Vec<TypeDefinition>,
    /// All builtin functions
    pub builtins: Vec<BuiltinDefinition>,
    /// Module definitions with their function mappings
    pub modules: Vec<ModuleDefinition>,
    /// Operators
    pub operators: Vec<OperatorDefinition>,
    /// Syntax patterns
    pub syntax: SyntaxPatterns,
    /// Categories of functionality
    pub categories: Vec<CategoryInfo>,
    /// OS operations ontology (platform-abstracted capabilities)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub os_ontology: Option<JsonValue>,
    /// CLI tool wrappers with install detection
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cli_tools: Option<JsonValue>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LanguageInfo {
    pub name: String,
    pub version: String,
    pub description: String,
    pub paradigms: Vec<String>,
    pub typing: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TypeDefinition {
    pub name: String,
    pub description: String,
    pub json_equivalent: String,
    pub examples: Vec<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub fields: Option<Vec<FieldDefinition>>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FieldDefinition {
    pub name: String,
    #[serde(rename = "type")]
    pub field_type: String,
    pub description: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BuiltinDefinition {
    pub name: String,
    pub description: String,
    pub category: String,
    pub signature: String,
    pub parameters: Vec<ParameterDefinition>,
    pub return_type: String,
    pub examples: Vec<ExampleDefinition>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub aliases: Option<Vec<String>>,
    /// Simplified JSON calling convention
    pub json_schema: JsonValue,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ParameterDefinition {
    pub name: String,
    #[serde(rename = "type")]
    pub param_type: String,
    pub description: String,
    pub required: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub default: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExampleDefinition {
    /// The example's own text. Empty (and left out of the output) when the
    /// example has none: repeating the builtin's description on every example
    /// cost date_diff ~13 KB for one paragraph said ten times.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub description: String,
    pub code: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub result: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OperatorDefinition {
    pub symbol: String,
    pub name: String,
    pub description: String,
    pub precedence: u8,
    pub associativity: String,
    pub operand_types: Vec<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SyntaxPatterns {
    pub variable_declaration: String,
    pub function_definition: String,
    pub lambda: String,
    pub pipeline: String,
    pub record_literal: String,
    pub array_literal: String,
    pub conditional: String,
    pub pattern_matching: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CategoryInfo {
    pub name: String,
    pub description: String,
    pub builtin_count: usize,
}

/// Module definition for the ontology — represents a namespace like `sys`, `net`, `file`
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ModuleDefinition {
    /// Module name (e.g., "sys", "net", "file")
    pub name: String,
    /// Human-readable description
    pub description: String,
    /// Functions exposed by this module
    pub functions: Vec<ModuleFunctionDef>,
}

/// A function within a module
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ModuleFunctionDef {
    /// Public method name (e.g., "hostname" in `sys.hostname()`)
    pub method: String,
    /// Internal builtin name it maps to
    pub builtin: String,
    /// Calling syntax
    pub syntax: String,
}

// ============================================================================
// API Implementation
// ============================================================================

/// Whether deterministic mode is on: `--deterministic` sets
/// `IRONSHELL_DETERMINISTIC`, and `repl::run_one` applies the same test before
/// it renders `ish -c` output as canonical JSON.
pub fn deterministic_mode() -> bool {
    std::env::var("IRONSHELL_DETERMINISTIC")
        .map(|s| s == "1" || s.eq_ignore_ascii_case("true"))
        .unwrap_or(false)
}

/// Metadata keys whose values change from run to run. The API sets none of
/// them today; they are dropped in deterministic mode so one added later
/// cannot break reproducibility.
const NONDETERMINISTIC_METADATA_KEYS: &[&str] = &[
    "execution_time_ms",
    "duration_ms",
    "elapsed_ms",
    "timestamp",
    "created_at",
    "started_at",
    "finished_at",
    "request_id",
    "id",
];

/// Rebuild every object under `v` with its keys in sorted order, so the
/// serialized bytes do not depend on the map implementation (serde_json's
/// `preserve_order` feature is off here, which already sorts, but this does not
/// rely on it).
fn sort_json_keys(v: &mut JsonValue) {
    match v {
        JsonValue::Object(m) => {
            let mut entries: Vec<(String, JsonValue)> = std::mem::take(m).into_iter().collect();
            entries.sort_by(|a, b| a.0.cmp(&b.0));
            for (_, child) in entries.iter_mut() {
                sort_json_keys(child);
            }
            *m = entries.into_iter().collect();
        }
        JsonValue::Array(items) => items.iter_mut().for_each(sort_json_keys),
        _ => {}
    }
}

/// Make a response reproducible under deterministic mode: canonical key order
/// in `result`, and no run-dependent metadata. Applied once, at the exit of
/// `process_request`, so every entry point (HTTP, WebSocket, stdio, library)
/// gets the same bytes.
fn deterministic_scrub(resp: &mut AgentResponse) {
    if let Some(result) = resp.result.as_mut() {
        sort_json_keys(result);
    }
    if let Some(JsonValue::Object(meta)) = resp.metadata.as_mut() {
        for key in NONDETERMINISTIC_METADATA_KEYS {
            meta.remove(*key);
        }
    }
}

/// Process an agent API request
pub fn process_request(request: &AgentRequest) -> AgentResponse {
    let mut resp = dispatch_request(request);
    annotate_error_fields(&mut resp);
    if deterministic_mode() {
        deterministic_scrub(&mut resp);
    }
    annotate_token_accounting(request, &mut resp);
    resp
}

fn dispatch_request(request: &AgentRequest) -> AgentResponse {
    match request {
        AgentRequest::Call { builtin, args } => execute_call(builtin, args),
        AgentRequest::Pipeline { steps, input } => execute_pipeline(steps, input.as_ref()),
        AgentRequest::Eval { code } => execute_eval(code),
        AgentRequest::Describe { builtin } => describe_builtin(builtin),
        AgentRequest::ListBuiltins { category } => list_builtins(category.as_deref()),
        AgentRequest::Schema { format, module, q } => {
            get_schema(format, module.as_deref(), q.as_deref())
        }
        AgentRequest::TypeInfo { type_name } => get_type_info(type_name.as_deref()),
    }
}

/// Attach estimated `token_accounting { tokens_in, tokens_out, tokens_total }`
/// to the response metadata (§6.1), so agents/supervisors get a budget signal
/// per call without changing the response struct's required fields.
fn annotate_token_accounting(request: &AgentRequest, resp: &mut AgentResponse) {
    let tokens_in =
        crate::builtins::est_token_count(&serde_json::to_string(request).unwrap_or_default());
    let out_str = match (&resp.result, &resp.error) {
        (Some(r), _) => serde_json::to_string(r).unwrap_or_default(),
        (None, Some(e)) => e.clone(),
        _ => String::new(),
    };
    let tokens_out = crate::builtins::est_token_count(&out_str);
    let acct = json!({
        "tokens_in": tokens_in,
        "tokens_out": tokens_out,
        "tokens_total": tokens_in + tokens_out,
        "estimated": true,
    });
    resp.metadata = Some(match resp.metadata.take() {
        Some(JsonValue::Object(mut m)) => {
            m.insert("token_accounting".to_string(), acct);
            JsonValue::Object(m)
        }
        Some(other) => json!({ "info": other, "token_accounting": acct }),
        None => json!({ "token_accounting": acct }),
    });
}

/// Copy the `code` and `hint` of a structured error into the response fields.
///
/// The `error` text is left as it was. Clients that read it still work, and the
/// fields are the same facts for clients that want to branch on them.
fn annotate_error_fields(resp: &mut AgentResponse) {
    if let Some((code, hint)) = resp.error.as_deref().and_then(structured_error_fields) {
        resp.code = Some(code);
        resp.hint = hint;
    }
}

/// The `code` and `hint` of the structured error that `error` carries, if any.
///
/// A failure reaches the API as `{"error":{"code":…,"hint":…}}`, either whole or
/// inside a message such as `Failed to build step 'where': {…}`. Each `{` in the
/// text is tried in turn, and the first object with an `error` that has a `code`
/// is the one reported.
fn structured_error_fields(error: &str) -> Option<(String, Option<String>)> {
    error.match_indices('{').find_map(|(start, _)| {
        let value = serde_json::Deserializer::from_str(&error[start..])
            .into_iter::<JsonValue>()
            .next()?
            .ok()?;
        let body = value.get("error")?;
        let code = body.get("code")?.as_str()?.to_string();
        let hint = body
            .get("hint")
            .and_then(JsonValue::as_str)
            .map(str::to_string);
        Some((code, hint))
    })
}

/// Execute a single builtin call
fn execute_call(builtin: &str, args: &JsonValue) -> AgentResponse {
    let started = std::time::Instant::now();
    // A module-qualified name (`math.sqrt`) calls the builtin that module member
    // is, as eval resolves it. Any other dotted name is refused, with the bare
    // builtin name as the hint.
    let resolved;
    let builtin = match builtin.split_once('.') {
        None => builtin,
        Some((module, member)) => match crate::modules::builtin_for_member(module, member) {
            Some(name) => {
                resolved = name;
                resolved.as_str()
            }
            None => {
                crate::record::note_outcome(
                    crate::record::Surface::Api,
                    "",
                    |_| vec![builtin.to_string()],
                    Err(crate::safety::ErrorCode::BadArg),
                    started.elapsed(),
                );
                return AgentResponse {
                    code: Some("E_BAD_ARG".to_string()),
                    hint: Some(format!(
                        "drop the module prefix and name the bare builtin: {{\"action\":\"call\",\"builtin\":\"{member}\"}}, or describe \"{member}\" first to check it is one"
                    )),
                    success: false,
                    result: None,
                    error: Some(format!(
                        "'{builtin}' is not a builtin name or a module member"
                    )),
                    result_type: None,
                    metadata: Some(json!({"code": "E_BAD_ARG"})),
                };
            }
        },
    };
    // A name that is no builtin is refused as one, before its arguments are
    // read: the argument check would otherwise name 'args' for a builtin that
    // is not there.
    if let Some(refusal) = unknown_builtin_refusal(builtin) {
        crate::record::note_outcome(
            crate::record::Surface::Api,
            "",
            |_| vec![builtin.to_string()],
            Err(crate::safety::ErrorCode::UnknownBuiltin),
            started.elapsed(),
        );
        return refusal;
    }
    let unwrapped = unwrap_args_envelope(builtin, args);
    let args = &unwrapped;
    // Convert JSON args to IronShell code
    let code = match args_to_code(builtin, args) {
        Ok(c) => c,
        Err(e) => {
            // No IronShell source was built, so the attempt is recorded under
            // its builtin name alone.
            crate::record::note_outcome(
                crate::record::Surface::Api,
                "",
                |_| vec![builtin.to_string()],
                Err(crate::safety::ErrorCode::BadArg),
                started.elapsed(),
            );
            return AgentResponse {
                code: Some("E_BAD_ARG".to_string()),
                hint: Some(
                    "pass the arguments as a JSON array in parameter order, or as an object keyed by the parameter names describe lists; a /call/:builtin body is that arguments object itself (e.g. {\"path\": \".\"}), not {\"args\": ...}"
                        .to_string(),
                ),
                success: false,
                result: None,
                error: Some(format!("Failed to convert arguments: {}", e)),
                result_type: None,
                metadata: Some(json!({"code": "E_BAD_ARG"})),
            };
        }
    };

    execute_eval(&code)
}

/// Execute a pipeline of operations
fn execute_pipeline(steps: &[PipelineStep], input: Option<&JsonValue>) -> AgentResponse {
    match run_pipeline_steps(steps, input, |_| {}) {
        Ok((value, code)) => response_from_value(value, &code),
        Err(failure) => AgentResponse {
            code: None,
            hint: None,
            success: false,
            result: None,
            error: Some(failure.message),
            result_type: None,
            metadata: failure
                .code_attempted
                .map(|code| json!({ "code_attempted": echo_code(&code) })),
        },
    }
}

/// A pipeline that did not produce a value.
struct PipelineFailure {
    /// 0-based index of the step that failed, when a step did.
    step: Option<usize>,
    message: String,
    /// The pipeline as one program, once it had been assembled.
    code_attempted: Option<String>,
    /// The failure's code, for the evaluation recorder. A step's own error
    /// keeps its code; a pipeline that could not be built is `E_BAD_ARG`.
    code: crate::safety::ErrorCode,
}

/// Name of the environment slot that carries stage `j`'s output to stage `j + 1`.
fn pipeline_slot(j: usize) -> String {
    format!("__pipeline_stage_{j}")
}

/// Parse `source` as exactly one expression statement.
fn single_expression(source: &str) -> Result<Expr, String> {
    // A parse failure is already the structured `E_PARSE` JSON, so it is
    // passed on as it is; prefixing it would make it unparseable for a client.
    let mut stmts = parse_program(source).map_err(|e| e.to_string())?;
    match (stmts.len(), stmts.pop()) {
        (1, Some(Stmt::Expr(expr))) => Ok(expr),
        _ => Err(
            crate::parser::ParseError::new("a pipeline stage must be a single expression")
                .to_string(),
        ),
    }
}

/// Run a pipeline one stage at a time, and return its value with the pipeline
/// as one program (the code it is reported under).
///
/// The stages are the optional input literal, then each step's code. Stage
/// `j > 0` runs as `slot(j-1) | code`, so each step gets the previous output
/// exactly as the joined program `input | step1 | step2` would pass it on.
/// Everything is built and parsed before the first stage runs, so a malformed
/// pipeline has no effects. `on_step_done(i)` runs after step `i` (0-based)
/// completes; a failure stops the pipeline at that step.
fn run_pipeline_steps(
    steps: &[PipelineStep],
    input: Option<&JsonValue>,
    on_step_done: impl FnMut(usize),
) -> Result<(Value, String), PipelineFailure> {
    // Every route through the runner is one attempt: `pipeline`, the
    // `execute` and `stream/pipeline` forms, and the WebSocket all come here.
    let started = std::time::Instant::now();
    let outcome = run_pipeline_steps_unrecorded(steps, input, on_step_done);
    match &outcome {
        Ok((_, code)) => crate::record::note_outcome(
            crate::record::Surface::Api,
            code,
            crate::record::heads_of,
            Ok(()),
            started.elapsed(),
        ),
        Err(failure) => crate::record::note_outcome(
            crate::record::Surface::Api,
            failure.code_attempted.as_deref().unwrap_or(""),
            crate::record::heads_of,
            Err(failure.code),
            started.elapsed(),
        ),
    }
    outcome
}

fn run_pipeline_steps_unrecorded(
    steps: &[PipelineStep],
    input: Option<&JsonValue>,
    mut on_step_done: impl FnMut(usize),
) -> Result<(Value, String), PipelineFailure> {
    use crate::safety::ErrorCode;

    if steps.is_empty() {
        return Err(PipelineFailure {
            step: None,
            message: "Pipeline must have at least one step".to_string(),
            code_attempted: None,
            code: ErrorCode::BadArg,
        });
    }

    let mut stage_codes: Vec<String> = Vec::with_capacity(steps.len() + 1);
    if let Some(input_val) = input {
        let literal = json_to_ae_literal(input_val).map_err(|e| PipelineFailure {
            step: None,
            message: format!("Failed to build pipeline input: {}", e),
            code_attempted: None,
            code: ErrorCode::BadArg,
        })?;
        stage_codes.push(literal);
    }
    for step in steps {
        let code = build_step_code(step).map_err(|e| PipelineFailure {
            step: None,
            message: format!("Failed to build step '{}': {}", step.builtin, e),
            code_attempted: None,
            code: ErrorCode::BadArg,
        })?;
        stage_codes.push(code);
    }

    let joined = stage_codes.join(" | ");
    if let Err(e) = parse_program(&joined) {
        return Err(PipelineFailure {
            step: None,
            message: e.to_string(),
            code_attempted: Some(joined),
            code: ErrorCode::BadArg,
        });
    }
    let mut stage_exprs = Vec::with_capacity(stage_codes.len());
    for (j, code) in stage_codes.iter().enumerate() {
        let source = if j == 0 {
            code.clone()
        } else {
            format!("{} | {}", pipeline_slot(j - 1), code)
        };
        match single_expression(&source) {
            Ok(expr) => stage_exprs.push(expr),
            Err(message) => {
                return Err(PipelineFailure {
                    step: None,
                    message,
                    code_attempted: Some(joined.clone()),
                    code: ErrorCode::BadArg,
                })
            }
        }
    }

    // Stage `j` is the step `j - offset`; the input literal, when present, is stage 0.
    let offset = usize::from(input.is_some());
    let last_stage = stage_exprs.len() - 1;
    let mut env = crate::modules::env_with_modules();
    let mut output = Value::Null;
    for (j, expr) in stage_exprs.into_iter().enumerate() {
        let step = j.checked_sub(offset);
        let fail = |message: String, code: ErrorCode| PipelineFailure {
            step,
            message,
            code_attempted: Some(joined.clone()),
            code,
        };
        // Statements start with no pipe input; so does each stage.
        env.set_input(None);
        let value = eval_expr(&expr, &mut env)
            .map_err(|e| fail(format!("{}", e), crate::safety::code_of(&e)))?;
        if let Some(i) = step {
            on_step_done(i);
        }
        if j < last_stage {
            env.set_var(pipeline_slot(j), value)
                .map_err(|message| fail(message, ErrorCode::Unknown))?;
        } else {
            output = value;
        }
    }

    // As eval_program does: an error value from the final expression is an uncaught error.
    if let Value::Error(msg) = &output {
        return Err(PipelineFailure {
            step: steps.len().checked_sub(1),
            message: format!("uncaught error: {msg}"),
            code_attempted: Some(joined),
            code: ErrorCode::Unknown,
        });
    }
    Ok((output, joined))
}

/// `value` as an agent may see it: secrets scrubbed in agent mode unless
/// `IRONSHELL_REDACT=off`, as `ish --agent` scrubs them. The Agent API and
/// MCP returned records and scalars as they were, so a password that
/// `ish --agent -c` hid came back in the clear over HTTP.
pub fn redact_for_agent(value: Value) -> Value {
    if crate::safety::current_mode() == crate::safety::Mode::Agent
        && crate::safety::redaction_enabled()
    {
        crate::builtins::redact_value(&value)
    } else {
        value
    }
}

/// The response for an evaluated value. The result is always JSON, so an Array,
/// a Record or a Table is an array or an object to the caller, never a string.
/// In agent mode a tabular result is paged to `IRONSHELL_TOKEN_BUDGET` when
/// that is set, or kept as a handle when it is too large to send whole.
fn response_from_value(value: Value, code: &str) -> AgentResponse {
    let value = redact_for_agent(value);
    let result_type = value_type_name(&value);
    // Deterministic output is the whole value in canonical form, so the
    // paging and handles of agent mode apply only when it is off.
    let paged = !deterministic_mode()
        && crate::safety::current_mode() == crate::safety::Mode::Agent
        && matches!(value, Value::Array(_) | Value::Table(_));
    let mut metadata = serde_json::Map::new();
    metadata.insert("code_executed".to_string(), json!(echo_code(code)));
    let result = if paged {
        match tabular_result(&value, &mut metadata) {
            Ok(result) => result,
            Err(e) => {
                return AgentResponse {
                    code: None,
                    hint: None,
                    success: false,
                    result: None,
                    error: Some(e.to_string()),
                    result_type: None,
                    metadata: Some(JsonValue::Object(metadata)),
                }
            }
        }
    } else {
        value_to_json(&value)
    };
    AgentResponse {
        code: None,
        hint: None,
        success: true,
        result: Some(result),
        error: None,
        result_type: Some(result_type),
        metadata: Some(JsonValue::Object(metadata)),
    }
}

/// Source code as metadata echoes it: whole when it is short, otherwise its
/// first 200 characters and its length. A caller that sent a large program
/// does not get it back in full.
fn echo_code(code: &str) -> String {
    let chars = code.chars().count();
    if chars < 1000 {
        code.to_string()
    } else {
        let head: String = code.chars().take(200).collect();
        format!("{head}... ({chars} chars)")
    }
}

/// Rows a handled result shows in its preview.
const HANDLE_PREVIEW_ROWS: usize = 3;

/// The JSON `result` for an Array or Table in agent mode. With
/// `IRONSHELL_TOKEN_BUDGET` set, the rows that fit the budget are returned and
/// the count left out is stated in `metadata`. Without a budget, a result too
/// large to send whole is stored as a handle (`metadata.handle`), and its first
/// rows are returned; `handle("<id>")` gives the whole value back.
fn tabular_result(
    value: &Value,
    metadata: &mut serde_json::Map<String, JsonValue>,
) -> Result<JsonValue> {
    let full = value_to_json(value);
    let total = table_rows(&full).len();
    let budget = std::env::var("IRONSHELL_TOKEN_BUDGET")
        .ok()
        .and_then(|s| s.parse::<usize>().ok())
        .filter(|m| *m > 0);
    if let Some(max) = budget {
        let shown = rows_within_budget(&full, max);
        if shown < total {
            metadata.insert("total".to_string(), json!(total));
            metadata.insert("shown".to_string(), json!(shown));
            metadata.insert("omitted".to_string(), json!(total - shown));
        }
        return Ok(first_rows(&full, shown));
    }
    let full_bytes = full.to_string().len();
    if crate::handles::worth_handling(value, full_bytes) {
        let shown = total.min(HANDLE_PREVIEW_ROWS);
        let handle = crate::handles::put(value.clone(), full_bytes)?;
        metadata.insert("handle".to_string(), json!(handle.id));
        metadata.insert("shape".to_string(), json!(handle.shape));
        metadata.insert("items".to_string(), json!(handle.items));
        metadata.insert("bytes".to_string(), json!(handle.bytes));
        metadata.insert("total".to_string(), json!(total));
        metadata.insert("shown".to_string(), json!(shown));
        metadata.insert("omitted".to_string(), json!(total - shown));
        return Ok(first_rows(&full, shown));
    }
    Ok(full)
}

/// The rows of an Array, or of a Table's JSON form.
fn table_rows(full: &JsonValue) -> &[JsonValue] {
    match full {
        JsonValue::Array(rows) => rows.as_slice(),
        JsonValue::Object(obj) => match obj.get("rows") {
            Some(JsonValue::Array(rows)) => rows.as_slice(),
            _ => &[],
        },
        _ => &[],
    }
}

/// How many leading rows fit in `max` estimated tokens. One row is always
/// shown, so a single large row still makes progress.
fn rows_within_budget(full: &JsonValue, max: usize) -> usize {
    let mut used = 0usize;
    let mut shown = 0usize;
    for row in table_rows(full) {
        let tokens = crate::builtins::est_token_count(&row.to_string());
        if shown > 0 && used.saturating_add(tokens) > max {
            break;
        }
        used = used.saturating_add(tokens);
        shown += 1;
    }
    shown
}

/// `full` with only its first `n` rows.
fn first_rows(full: &JsonValue, n: usize) -> JsonValue {
    match full {
        JsonValue::Array(rows) => JsonValue::Array(rows.iter().take(n).cloned().collect()),
        JsonValue::Object(obj) => {
            let mut obj = obj.clone();
            let page: Vec<JsonValue> = table_rows(full).iter().take(n).cloned().collect();
            obj.insert("rows".to_string(), JsonValue::Array(page));
            JsonValue::Object(obj)
        }
        other => other.clone(),
    }
}

/// Execute raw IronShell code
fn execute_eval(code: &str) -> AgentResponse {
    // Arbitrary source: the module namespaces must be bound, or every
    // `file.read(…)`/`sys.hostname()` in it fails as a field access on Null.
    let mut env = crate::modules::env_with_modules();

    let started = std::time::Instant::now();
    let outcome = parse_program(code).and_then(|stmts| eval_program(&stmts, &mut env));
    crate::record::note(
        crate::record::Surface::Api,
        code,
        &outcome,
        started.elapsed(),
    );
    match outcome {
        Ok(value) => response_from_value(value, code),
        // A parse failure is the structured `E_PARSE` JSON, as the CLI reports it.
        Err(e) => AgentResponse {
            code: None,
            hint: None,
            success: false,
            result: None,
            error: Some(e.to_string()),
            result_type: None,
            metadata: Some(json!({
                "code_attempted": echo_code(code)
            })),
        },
    }
}

/// The description of a manifest builtin of a loaded plugin. The manifest
/// declares no signature, parameters or examples, so the entry says only that
/// the name exists, that its body is arbitrary IronShell, and the effect the
/// dispatch gate applies to it (`Exec`, which agent mode approves per call).
fn plugin_builtin_response(name: &str) -> AgentResponse {
    let definition = BuiltinDefinition {
        name: name.to_string(),
        description: "A builtin provided by a loaded plugin; its body is arbitrary IronShell"
            .to_string(),
        category: "Plugin".to_string(),
        signature: format!("{name}(...)"),
        parameters: Vec::new(),
        return_type: "Any".to_string(),
        examples: Vec::new(),
        aliases: None,
        json_schema: json!({"type": "object"}),
    };
    AgentResponse {
        code: None,
        hint: None,
        success: true,
        result: Some(serde_json::to_value(&definition).unwrap_or(JsonValue::Null)),
        error: None,
        result_type: Some("BuiltinDefinition".to_string()),
        metadata: Some(json!({"effect": crate::safety::effect_of(name).as_str()})),
    }
}

/// Get information about a specific builtin
fn describe_builtin(name: &str) -> AgentResponse {
    // `file.read` names the builtin that member runs (`file_read`), as eval and
    // `call` resolve it.
    let resolved = name
        .split_once('.')
        .and_then(|(module, member)| crate::modules::builtin_for_member(module, member));
    let name = resolved.as_deref().unwrap_or(name);
    // The whole catalogue, as the ontology serves it. This read the 16
    // hand-written entries, so `/api/v1/builtins/ontology_manifest` answered
    // "not found" for a builtin the schema lists.
    let builtins = get_all_builtin_definitions();

    if let Some(builtin) = builtins.iter().find(|b| b.name == name) {
        AgentResponse {
            code: None,
            hint: None,
            success: true,
            result: Some(serde_json::to_value(builtin).unwrap_or(JsonValue::Null)),
            error: None,
            result_type: Some("BuiltinDefinition".to_string()),
            metadata: None,
        }
    } else if crate::plugins::is_manifest_builtin(name) {
        plugin_builtin_response(name)
    } else {
        // Check for aliases
        if let Some(builtin) = builtins.iter().find(|b| {
            b.aliases
                .as_ref()
                .map(|a| a.contains(&name.to_string()))
                .unwrap_or(false)
        }) {
            AgentResponse {
                code: None,
                hint: None,
                success: true,
                result: Some(serde_json::to_value(builtin).unwrap_or(JsonValue::Null)),
                error: None,
                result_type: Some("BuiltinDefinition".to_string()),
                metadata: Some(
                    json!({"note": format!("'{}' is an alias for '{}'", name, builtin.name)}),
                ),
            }
        } else {
            let did_you_mean =
                crate::builtins::nearest_names(name, builtins.iter().map(|b| b.name.as_str()));
            let hint = if did_you_mean.is_empty() {
                "search by keyword with suggest(\"<words>\"), or list a category with ontology_manifest()".to_string()
            } else {
                format!("did you mean {}?", did_you_mean.join(", "))
            };
            AgentResponse {
                code: Some("E_UNKNOWN_BUILTIN".to_string()),
                hint: Some(hint.clone()),
                success: false,
                result: None,
                error: Some(format!("Builtin '{}' not found; {}", name, hint)),
                result_type: None,
                metadata: Some(json!({"code": "E_UNKNOWN_BUILTIN", "did_you_mean": did_you_mean})),
            }
        }
    }
}

/// List all available builtins
fn list_builtins(category: Option<&str>) -> AgentResponse {
    let builtins = get_all_builtin_definitions();

    let filtered: Vec<_> = if let Some(cat) = category {
        builtins
            .into_iter()
            .filter(|b| b.category.to_lowercase() == cat.to_lowercase())
            .collect()
    } else {
        builtins
    };

    let summary: Vec<JsonValue> = filtered
        .iter()
        .map(|b| {
            json!({
                "name": b.name,
                "description": b.description,
                "category": b.category,
                "signature": b.signature
            })
        })
        .collect();

    AgentResponse {
        code: None,
        hint: None,
        success: true,
        result: Some(json!({
            "count": summary.len(),
            "builtins": summary
        })),
        error: None,
        result_type: Some("BuiltinList".to_string()),
        metadata: None,
    }
}

/// What a schema's bytes depend on besides its format: the surface and
/// deterministic-mode settings (`current_mode`, `deterministic_mode`), the
/// environment the CLI-tool probe reads (`PATH`, `PATHEXT`, the working
/// directory), and the filter. A change to any of them selects a different
/// cache entry, so a filtered answer is never served for an unfiltered request.
type SchemaCacheKey = (String, String, bool, String, String, String, String);

/// Built schema values, by format, environment and filter. The schema is a
/// function of those inputs, and building it probes the PATH for CLI tools and
/// rebuilds the whole catalogue, so each one is built once per key rather than
/// per request. The filter is free text, so the map is bounded: a client that
/// varies `q` cannot grow it without limit.
static SCHEMA_CACHE: OnceLock<Mutex<HashMap<SchemaCacheKey, JsonValue>>> = OnceLock::new();

/// Most entries the schema cache holds before it starts again.
const SCHEMA_CACHE_LIMIT: usize = 256;

/// A filter string as the schema uses it: trimmed, lower-cased, and absent when
/// empty. Matching is case-insensitive, so `File` and `file` are one request.
fn normalise_filter(value: Option<&str>) -> Option<String> {
    value
        .map(|v| v.trim().to_lowercase())
        .filter(|v| !v.is_empty())
}

/// Get the language schema in the requested format, narrowed by `module` and
/// `q` when either is given.
fn get_schema(format: &SchemaFormat, module: Option<&str>, q: Option<&str>) -> AgentResponse {
    let module = normalise_filter(module);
    let q = normalise_filter(q);
    let key: SchemaCacheKey = (
        format!("{:?}", format),
        format!("{:?}", crate::safety::current_mode()),
        deterministic_mode(),
        std::env::var("PATH").unwrap_or_default(),
        std::env::var("PATHEXT").unwrap_or_default(),
        std::env::current_dir()
            .map(|p| p.display().to_string())
            .unwrap_or_default(),
        format!("module={module:?} q={q:?}"),
    );
    let cache = SCHEMA_CACHE.get_or_init(|| Mutex::new(HashMap::new()));
    let cached = cache
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
        .get(&key)
        .cloned();
    let result = match cached {
        Some(value) => value,
        None => {
            // Only a request that is accepted is cached, so an unknown module
            // is refused on every ask rather than remembered.
            let mut ontology = build_language_ontology();
            if let Err(refused) = narrow_ontology(&mut ontology, module.as_deref(), q.as_deref()) {
                return refused;
            }
            let built = build_schema_value(format, &ontology);
            let mut cache_map = cache
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            if cache_map.len() >= SCHEMA_CACHE_LIMIT {
                cache_map.clear();
            }
            cache_map.insert(key, built.clone());
            built
        }
    };

    let filter = (module.is_some() || q.is_some()).then(|| json!({"module": module, "q": q}));
    AgentResponse {
        code: None,
        hint: None,
        success: true,
        result: Some(result),
        error: None,
        result_type: Some(format!("{:?}Schema", format)),
        metadata: filter.map(|f| json!({"filter": f})),
    }
}

/// Narrow the catalogue to the builtins in `module` whose name or description
/// contains `q`. This runs before any provider format is built, so every format
/// shrinks alike. A module is kept only with the functions that survive, and a
/// module with none left is dropped. An unknown module is refused, with the
/// modules it might have meant.
fn narrow_ontology(
    ontology: &mut LanguageOntology,
    module: Option<&str>,
    q: Option<&str>,
) -> Result<(), AgentResponse> {
    if module.is_none() && q.is_none() {
        return Ok(());
    }
    if let Some(m) = module {
        if !ontology
            .modules
            .iter()
            .any(|md| md.name.eq_ignore_ascii_case(m))
        {
            let did_you_mean = crate::builtins::nearest_names(
                m,
                ontology.modules.iter().map(|md| md.name.as_str()),
            );
            let hint = if did_you_mean.is_empty() {
                "ontology_manifest() lists every module".to_string()
            } else {
                format!(
                    "did you mean {}? ontology_manifest() lists every module",
                    did_you_mean.join(", ")
                )
            };
            return Err(AgentResponse {
                code: Some("E_BAD_ARG".to_string()),
                hint: Some(hint.clone()),
                success: false,
                result: None,
                error: Some(format!("Unknown module '{m}'; {hint}")),
                result_type: None,
                metadata: Some(json!({"code": "E_BAD_ARG", "did_you_mean": did_you_mean})),
            });
        }
    }

    let in_module: Option<Vec<String>> = module.map(|m| {
        ontology
            .modules
            .iter()
            .filter(|md| md.name.eq_ignore_ascii_case(m))
            .flat_map(|md| md.functions.iter().map(|f| f.builtin.clone()))
            .collect()
    });
    // A module member names the implementation (`file_delete_lines`), and one
    // implementation can be listed under its alias (`delete_lines`). The entry
    // is kept under either name, so `file` lists `delete_lines`.
    let member_of = |b: &BuiltinDefinition, names: &[String]| {
        names.contains(&b.name)
            || b.aliases
                .as_ref()
                .is_some_and(|a| a.iter().any(|x| names.contains(x)))
    };
    ontology.builtins.retain(|b| {
        in_module
            .as_ref()
            .is_none_or(|names| member_of(b, names.as_slice()))
            && q.is_none_or(|q| {
                b.name.to_lowercase().contains(q) || b.description.to_lowercase().contains(q)
            })
    });

    let kept: HashSet<&str> = ontology
        .builtins
        .iter()
        .flat_map(|b| {
            std::iter::once(b.name.as_str()).chain(b.aliases.iter().flatten().map(String::as_str))
        })
        .collect();
    for md in &mut ontology.modules {
        md.functions.retain(|f| kept.contains(f.builtin.as_str()));
    }
    ontology.modules.retain(|md| !md.functions.is_empty());
    Ok(())
}

/// Build the schema value for `format` from `ontology`, without the cache.
fn build_schema_value(format: &SchemaFormat, ontology: &LanguageOntology) -> JsonValue {
    match format {
        SchemaFormat::JsonSchema => serde_json::to_value(&ontology).unwrap_or(JsonValue::Null),
        SchemaFormat::OpenAI => build_openai_schema(&ontology),
        SchemaFormat::Claude => build_claude_schema(&ontology),
        SchemaFormat::Gemini => build_gemini_schema(&ontology),
        SchemaFormat::Llama => build_llama_schema(&ontology),
        SchemaFormat::Mistral => build_mistral_schema(&ontology),
        SchemaFormat::Cohere => build_cohere_schema(&ontology),
        SchemaFormat::Grok => build_grok_schema(&ontology),
        SchemaFormat::DeepSeek => build_deepseek_schema(&ontology),
        SchemaFormat::Bedrock => build_bedrock_schema(&ontology),
        SchemaFormat::AzureOpenAI => build_azure_openai_schema(&ontology),
        SchemaFormat::Qwen => build_qwen_schema(&ontology),
        SchemaFormat::Ollama => build_ollama_schema(&ontology),
        SchemaFormat::VLLM => build_vllm_schema(&ontology),
        SchemaFormat::HuggingFace => build_huggingface_schema(&ontology),
        SchemaFormat::OpenRouter => build_openrouter_schema(&ontology),
        SchemaFormat::Kimi => build_kimi_schema(&ontology),
        SchemaFormat::Yi => build_yi_schema(&ontology),
        SchemaFormat::GLM => build_glm_schema(&ontology),
        SchemaFormat::Reka => build_reka_schema(&ontology),
        SchemaFormat::AI21 => build_ai21_schema(&ontology),
        SchemaFormat::Perplexity => build_perplexity_schema(&ontology),
        SchemaFormat::Together => build_together_schema(&ontology),
        SchemaFormat::Groq => build_groq_schema(&ontology),
        SchemaFormat::Fireworks => build_fireworks_schema(&ontology),
        SchemaFormat::Ontology => build_compact_ontology(&ontology),
        SchemaFormat::JsonLD => crate::providers::ontology::OS_ONTOLOGY.to_json_ld(),
        SchemaFormat::OwlTurtle => {
            json!({"format": "text/turtle", "content": crate::providers::ontology::OS_ONTOLOGY.to_owl_turtle()})
        }
        SchemaFormat::SHACL => crate::providers::ontology::OS_ONTOLOGY.to_shacl(),
    }
}

/// Get type information
fn get_type_info(type_name: Option<&str>) -> AgentResponse {
    let types = get_type_definitions();

    if let Some(name) = type_name {
        if let Some(type_def) = types
            .iter()
            .find(|t| t.name.to_lowercase() == name.to_lowercase())
        {
            AgentResponse {
                code: None,
                hint: None,
                success: true,
                result: Some(serde_json::to_value(type_def).unwrap_or(JsonValue::Null)),
                error: None,
                result_type: Some("TypeDefinition".to_string()),
                metadata: None,
            }
        } else {
            AgentResponse {
                code: Some("E_NOT_FOUND".to_string()),
                hint: Some(
                    "send a type_info request without type_name to list the types".to_string(),
                ),
                success: false,
                result: None,
                error: Some(format!("Type '{}' not found", name)),
                result_type: None,
                metadata: None,
            }
        }
    } else {
        AgentResponse {
            code: None,
            hint: None,
            success: true,
            result: Some(serde_json::to_value(&types).unwrap_or(JsonValue::Null)),
            error: None,
            result_type: Some("TypeList".to_string()),
            metadata: None,
        }
    }
}

// ============================================================================
// Helper Functions
// ============================================================================

/// Convert the arguments of a `call` (or a pipeline step) to IronShell code
/// that calls `builtin`.
///
/// An array is positional and is written as it stands. An object is named, and
/// each name is mapped to a declared parameter by `named_args_to_positional`,
/// so no value is dropped. The arguments reach the code only through
/// `json_to_ae_literal`, which writes every string as a literal that evaluates
/// to that string.
fn args_to_code(builtin: &str, args: &JsonValue) -> Result<String> {
    if !is_identifier(builtin) {
        return Err(anyhow!(
            "'{}' is not a builtin name; a builtin name is letters, digits and underscores",
            builtin
        ));
    }
    let positional: Vec<JsonValue> = match args {
        JsonValue::Null => Vec::new(),
        JsonValue::Array(arr) => arr.clone(),
        JsonValue::Object(obj) => named_args_to_positional(builtin, obj)?,
        other => vec![other.clone()],
    };
    let literals = positional
        .iter()
        .map(json_to_ae_literal)
        .collect::<Result<Vec<_>>>()?;
    Ok(format!("{}({})", builtin, literals.join(", ")))
}

/// Map named arguments to the positions of `builtin`'s declared parameters, as
/// `get_all_builtin_definitions()` lists them.
///
/// A name that is not a parameter is refused, and the error lists the valid
/// names. A parameter that is left out must be optional, and no later one may
/// be given, since an omitted middle argument has no literal. A variadic last
/// parameter given an array is spread over the positions.
///
/// A builtin with a piped subject takes that subject as its first argument, so
/// in this form the subject is the first position. It is named `subject` (or
/// `input`) and its declared parameters follow it.
pub(crate) fn named_args_to_positional(
    builtin: &str,
    obj: &serde_json::Map<String, JsonValue>,
) -> Result<Vec<JsonValue>> {
    if obj.is_empty() {
        return Ok(Vec::new());
    }
    let definition = get_all_builtin_definitions().into_iter().find(|d| {
        d.name == builtin
            || d.aliases
                .as_ref()
                .is_some_and(|names| names.iter().any(|n| n == builtin))
    });
    let (canonical, params) = match definition {
        Some(d) => (d.name, d.parameters),
        None => (builtin.to_string(), Vec::new()),
    };
    let signature = crate::signature::signature_of(&canonical);
    let has_subject = signature.is_some_and(|s| s.subject.is_some());

    // The positions, in order: the subject first when there is one, then the
    // declared parameters. Each is a name and whether it must be given.
    // A definition built from the signature already lists the subject first.
    let mut positions: Vec<(String, bool)> = Vec::new();
    if has_subject && params.first().is_none_or(|p| p.name != "subject") {
        let required = signature.is_some_and(|s| s.subject_required);
        positions.push(("subject".to_string(), required));
    }
    positions.extend(params.iter().map(|p| (p.name.clone(), p.required)));

    let valid = if positions.is_empty() {
        "none (pass its arguments as an array)".to_string()
    } else {
        positions
            .iter()
            .map(|(name, _)| name.as_str())
            .collect::<Vec<_>>()
            .join(", ")
    };
    let mut slots: Vec<Option<JsonValue>> = vec![None; positions.len()];
    for (key, value) in obj {
        // `input` names the subject too, when the builtin has one and no
        // parameter of its own is called that.
        let index = positions
            .iter()
            .position(|(name, _)| name == key)
            .or_else(|| (has_subject && key == "input").then_some(0));
        let Some(i) = index else {
            return Err(anyhow!(
                "unknown argument '{}' for '{}'; valid parameters: {}",
                key,
                builtin,
                valid
            ));
        };
        if slots[i].is_some() {
            return Err(anyhow!(
                "argument '{}' of '{}' is given twice",
                positions[i].0,
                builtin
            ));
        }
        slots[i] = Some(value.clone());
    }

    if let Some((name, _)) = positions
        .iter()
        .zip(&slots)
        .find(|((_, required), slot)| *required && slot.is_none())
        .map(|(position, _)| position)
    {
        // `subject` is the key the args object takes the value the builtin works
        // on (what a pipeline would pipe in), so it is the name to give; say
        // what it is and list every key, since no declared parameter is called
        // that. Naming a declared parameter instead would be wrong for a
        // builtin whose subject comes before its first parameter (date_add's
        // date, before `amount`).
        if name == "subject" {
            return Err(anyhow!(
                "missing required argument 'subject' for '{}': the value it works on, which a pipeline would pipe in (the key 'input' is accepted too); valid parameters: {}",
                builtin,
                valid
            ));
        }
        return Err(anyhow!(
            "missing required argument '{}' for '{}'",
            name,
            builtin
        ));
    }

    let Some(last) = slots.iter().rposition(Option::is_some) else {
        return Ok(Vec::new());
    };
    let variadic_last = signature.is_some_and(|s| s.params.last().is_some_and(|p| p.variadic));
    let mut out = Vec::with_capacity(last + 1);
    for (i, (slot, (name, _))) in slots.into_iter().zip(&positions).enumerate() {
        if i > last {
            break;
        }
        match slot {
            Some(JsonValue::Array(items)) if variadic_last && i + 1 == positions.len() => {
                out.extend(items)
            }
            Some(value) => out.push(value),
            // A lone query: `{"source_or_query": "select 1"}` is `sql("select 1")`,
            // whose one String is the query, so the empty subject has no position.
            None if i == 0
                && name.as_str() == "subject"
                && matches!(canonical.as_str(), "sql" | "sql_value") => {}
            None => {
                return Err(anyhow!(
                    "argument '{}' of '{}' is optional, but a later argument is given; \
                     give '{}' a value too",
                    name,
                    builtin,
                    name
                ))
            }
        }
    }
    Ok(out)
}

/// The refusal for a name that is no builtin (`E_UNKNOWN_BUILTIN`, with the
/// nearest real names), or `None` when it names one. A module-qualified name
/// is resolved first, as `describe` resolves it.
pub fn unknown_builtin_refusal(name: &str) -> Option<AgentResponse> {
    let response = describe_builtin(name);
    (response.code.as_deref() == Some("E_UNKNOWN_BUILTIN")).then_some(response)
}

/// The arguments of a `/call/:builtin` body that was wrapped as the `/execute`
/// request is: `{"args": {...}}` or `{"args": [...]}`. When that is the whole
/// body and the builtin has no parameter named `args`, the wrapper is removed.
/// Anything else is returned as it came.
fn unwrap_args_envelope(builtin: &str, args: &JsonValue) -> JsonValue {
    let inner = match args {
        JsonValue::Object(obj) if obj.len() == 1 => match obj.get("args") {
            Some(inner @ (JsonValue::Object(_) | JsonValue::Array(_))) => inner,
            _ => return args.clone(),
        },
        _ => return args.clone(),
    };
    let takes_args = get_all_builtin_definitions()
        .into_iter()
        .find(|d| {
            d.name == builtin
                || d.aliases
                    .as_ref()
                    .is_some_and(|names| names.iter().any(|n| n == builtin))
        })
        .is_some_and(|d| d.parameters.iter().any(|p| p.name == "args"));
    if takes_args {
        args.clone()
    } else {
        inner.clone()
    }
}

/// The `code` of an `/eval` body: a string, or a coded `E_BAD_ARG` response
/// that names the field. A missing `code` used to evaluate the empty program
/// and answer `success: true, result: null`.
pub fn eval_code_from_body(body: &JsonValue) -> Result<String, AgentResponse> {
    match body.get("code") {
        Some(JsonValue::String(code)) => Ok(code.clone()),
        other => {
            let problem = match other {
                None => "is missing".to_string(),
                Some(v) => format!("must be a string, got {}", json_kind(v)),
            };
            Err(AgentResponse {
                code: Some("E_BAD_ARG".to_string()),
                hint: Some(
                    "send the IronShell source as a string: {\"code\": \"1 + 1\"}".to_string(),
                ),
                success: false,
                result: None,
                error: Some(format!("the request field 'code' {problem}")),
                result_type: None,
                metadata: Some(json!({"code": "E_BAD_ARG", "field": "code"})),
            })
        }
    }
}

fn json_kind(v: &JsonValue) -> &'static str {
    match v {
        JsonValue::Null => "null",
        JsonValue::Bool(_) => "a boolean",
        JsonValue::Number(_) => "a number",
        JsonValue::String(_) => "a string",
        JsonValue::Array(_) => "an array",
        JsonValue::Object(_) => "an object",
    }
}

/// Whether `name` is written as an IronShell identifier: a letter or an
/// underscore, then letters, digits or underscores.
fn is_identifier(name: &str) -> bool {
    let mut chars = name.chars();
    chars
        .next()
        .is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
        && chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
}

/// Build code for a single pipeline step
fn build_step_code(step: &PipelineStep) -> Result<String> {
    let mut code = args_to_code(&step.builtin, &step.args)?;

    // Handle predicate (for where/filter operations). The predicate is an
    // expression over `x` that is written into the code as text, so it must
    // parse alone as exactly one lambda. Anything more would close the call
    // early and add stages. Once it is written, it runs through the same
    // evaluator and effect gate as `/api/v1/eval`.
    if let Some(pred) = &step.predicate {
        // Parse simple predicates like "size > 1000"
        let lambda = predicate_to_lambda(pred);
        if !matches!(single_expression(&lambda), Ok(Expr::Lambda { .. })) {
            return Err(crate::safety::bad_arg(
                &step.builtin,
                "a predicate that is one expression over x",
                &format!("'{pred}'"),
            ));
        }
        code = format!("{}({})", step.builtin, lambda);
    }

    // Handle field selection. The field is written as `x.<field>`, so it must
    // be an identifier, or it could close the call and add stages.
    if let Some(field) = &step.select {
        if !is_identifier(field) {
            return Err(crate::safety::bad_arg(
                &step.builtin,
                "a field name of letters, digits and underscores",
                &format!("'{field}'"),
            ));
        }
        code = format!("{} | map(fn(x) => x.{})", code, field);
    }

    Ok(code)
}

/// Convert a simple predicate to a lambda
fn predicate_to_lambda(predicate: &str) -> String {
    // Handle common patterns
    if predicate.contains(">")
        || predicate.contains("<")
        || predicate.contains("==")
        || predicate.contains("!=")
    {
        format!("fn(x) => x.{}", predicate)
    } else if predicate.starts_with('.') {
        format!("fn(x) => x{}", predicate)
    } else {
        format!("fn(x) => {}", predicate)
    }
}

/// Convert a JSON value to IronShell literal syntax.
///
/// The result is code, so every piece of it must be a literal. A record key must
/// be an identifier, because IronShell has no quoted record keys; any other key
/// is refused rather than written into the code.
fn json_to_ae_literal(val: &JsonValue) -> Result<String> {
    Ok(match val {
        JsonValue::Null => "null".to_string(),
        JsonValue::Bool(b) => if *b { "true" } else { "false" }.to_string(),
        JsonValue::Number(n) => n.to_string(),
        JsonValue::String(s) => ae_string_literal(s),
        JsonValue::Array(arr) => {
            let items = arr
                .iter()
                .map(json_to_ae_literal)
                .collect::<Result<Vec<_>>>()?;
            format!("[{}]", items.join(", "))
        }
        JsonValue::Object(obj) => {
            let mut fields = Vec::with_capacity(obj.len());
            for (key, value) in obj {
                if !is_identifier(key) {
                    return Err(anyhow!(
                        "record key '{}' is not an identifier, so it cannot be written as a literal",
                        key
                    ));
                }
                fields.push(format!("{}: {}", key, json_to_ae_literal(value)?));
            }
            format!("{{{}}}", fields.join(", "))
        }
    })
}

/// A string as an IronShell string literal that evaluates to exactly `s`.
///
/// The lexer reads `\n`, `\r`, `\t`, `\\` and `\"` as escapes. The evaluator
/// also reads `${expr}` inside any string as an expression to run, so a `${`
/// in `s` is written as two literals joined with `+`, and no literal in the
/// code contains `${`.
fn ae_string_literal(s: &str) -> String {
    let mut body = String::with_capacity(s.len() + 2);
    for c in s.chars() {
        match c {
            '\\' => body.push_str("\\\\"),
            '"' => body.push_str("\\\""),
            '\n' => body.push_str("\\n"),
            '\r' => body.push_str("\\r"),
            '\t' => body.push_str("\\t"),
            c => body.push(c),
        }
    }
    if body.contains("${") {
        format!("(\"{}\")", body.replace("${", "$\" + \"{"))
    } else {
        format!("\"{}\"", body)
    }
}

/// Convert IronShell Value to JSON
fn value_to_json(value: &Value) -> JsonValue {
    match value {
        Value::Null => JsonValue::Null,
        Value::Bool(b) => json!(*b),
        Value::Int(i) => json!(*i),
        Value::Float(f) => json!(*f),
        Value::Str(s) => json!(s),
        Value::Uri(u) => json!({"_type": "Uri", "value": u}),
        Value::Array(arr) => {
            let items: Vec<JsonValue> = arr.iter().map(value_to_json).collect();
            JsonValue::Array(items)
        }
        Value::Record(fields) => {
            let obj: serde_json::Map<String, JsonValue> = fields
                .iter()
                .map(|(k, v)| (k.clone(), value_to_json(v)))
                .collect();
            JsonValue::Object(obj)
        }
        Value::Table(table) => {
            json!({
                "_type": "Table",
                "schema": table.schema,
                "rows": table.rows.iter().map(|row| {
                    let obj: serde_json::Map<String, JsonValue> = row
                        .iter()
                        .map(|(k, v)| (k.clone(), value_to_json(v)))
                        .collect();
                    JsonValue::Object(obj)
                }).collect::<Vec<_>>()
            })
        }
        Value::Lambda(lambda) => {
            json!({
                "_type": "Lambda",
                "params": lambda.params,
                "body": format!("{:?}", lambda.body)
            })
        }
        Value::AsyncLambda(lambda) => {
            json!({
                "_type": "AsyncLambda",
                "params": lambda.params,
                "body": format!("{:?}", lambda.body)
            })
        }
        Value::Future(future) => {
            json!({
                "_type": "Future",
                "args": future.args.iter().map(value_to_json).collect::<Vec<_>>()
            })
        }
        Value::Error(e) => json!({
            "_type": "Error",
            "message": e
        }),
        Value::Builtin(b) => json!({
            "_type": "Builtin",
            "name": b.name
        }),
    }
}

/// Get the type name for a Value
fn value_type_name(value: &Value) -> String {
    value.type_name().to_string()
}

// ============================================================================
// Schema Builders
// ============================================================================

/// Build the complete language ontology
pub fn build_language_ontology() -> LanguageOntology {
    LanguageOntology {
        ontology_version: "1.0.0".to_string(),
        language: LanguageInfo {
            name: "IronShell".to_string(),
            version: env!("CARGO_PKG_VERSION").to_string(),
            description: "A typed functional shell with multi-modal AI capabilities".to_string(),
            paradigms: vec![
                "Functional".to_string(),
                "Pipeline-oriented".to_string(),
                "Expression-based".to_string(),
            ],
            typing: "Hindley-Milner type inference".to_string(),
        },
        types: get_type_definitions(),
        builtins: get_all_builtin_definitions(),
        modules: get_module_definitions(),
        operators: get_operator_definitions(),
        syntax: SyntaxPatterns {
            variable_declaration: "let <name> = <expr>".to_string(),
            function_definition: "fn <name>(<params>) => <expr>".to_string(),
            lambda: "fn(<params>) => <expr>".to_string(),
            pipeline: "<expr> | <func> | <func>".to_string(),
            record_literal: "{ field1: value1, field2: value2 }".to_string(),
            array_literal: "[item1, item2, item3]".to_string(),
            conditional: "if <cond> then <expr> else <expr>".to_string(),
            pattern_matching: "match <expr> { pattern => result, ... }".to_string(),
        },
        categories: get_category_info(),
        os_ontology: Some(crate::providers::ontology::OS_ONTOLOGY.to_json_schema()),
        cli_tools: Some(crate::providers::ontology::CLI_TOOL_REGISTRY.to_json_schema()),
    }
}

/// Get type definitions
fn get_type_definitions() -> Vec<TypeDefinition> {
    vec![
        TypeDefinition {
            name: "Int".to_string(),
            description: "64-bit signed integer".to_string(),
            json_equivalent: "number (integer)".to_string(),
            examples: vec!["42".to_string(), "-17".to_string(), "0".to_string()],
            fields: None,
        },
        TypeDefinition {
            name: "Float".to_string(),
            description: "64-bit floating point number".to_string(),
            json_equivalent: "number".to_string(),
            examples: vec!["3.14".to_string(), "-0.5".to_string(), "1e10".to_string()],
            fields: None,
        },
        TypeDefinition {
            name: "String".to_string(),
            description: "UTF-8 text string".to_string(),
            json_equivalent: "string".to_string(),
            examples: vec!["\"hello\"".to_string(), "\"multi\\nline\"".to_string()],
            fields: None,
        },
        TypeDefinition {
            name: "Bool".to_string(),
            description: "Boolean true/false".to_string(),
            json_equivalent: "boolean".to_string(),
            examples: vec!["true".to_string(), "false".to_string()],
            fields: None,
        },
        TypeDefinition {
            name: "Array".to_string(),
            description: "Ordered collection of values".to_string(),
            json_equivalent: "array".to_string(),
            examples: vec!["[1, 2, 3]".to_string(), "[\"a\", \"b\"]".to_string()],
            fields: None,
        },
        TypeDefinition {
            name: "Record".to_string(),
            description: "Key-value mapping (like JSON object)".to_string(),
            json_equivalent: "object".to_string(),
            examples: vec!["{name: \"file.txt\", size: 1024}".to_string()],
            fields: None,
        },
        TypeDefinition {
            name: "Table".to_string(),
            description: "Tabular data with headers and rows".to_string(),
            json_equivalent: "object with headers and rows arrays".to_string(),
            examples: vec!["Result of ls(), csv parsing".to_string()],
            fields: Some(vec![
                FieldDefinition {
                    name: "headers".to_string(),
                    field_type: "Array<String>".to_string(),
                    description: "Column names".to_string(),
                },
                FieldDefinition {
                    name: "rows".to_string(),
                    field_type: "Array<Array<Value>>".to_string(),
                    description: "Row data".to_string(),
                },
            ]),
        },
        TypeDefinition {
            name: "Lambda".to_string(),
            description: "Anonymous function".to_string(),
            json_equivalent: "object with params and body".to_string(),
            examples: vec![
                "fn(x) => x * 2".to_string(),
                "fn(a, b) => a + b".to_string(),
            ],
            fields: None,
        },
        TypeDefinition {
            name: "Option".to_string(),
            description: "Optional value (Some or None)".to_string(),
            json_equivalent: "value or null".to_string(),
            examples: vec!["Some(42)".to_string(), "None".to_string()],
            fields: None,
        },
    ]
}

/// Annotate each builtin definition with its safety effect class (`pure`,
/// `read_local`, `write_local`, `destructive`, `process`, `network`, `exec`,
/// `privileged`) under the `x-effect` key of its `json_schema`, so an agent can
/// see an operation's danger level from the ontology *before* calling it.
/// See `crate::safety::effect_of`.
///
/// Also attaches `x-returns` — the builtin's proven return shape — where one
/// exists. Together these say what a call *costs* and what it *yields* before
/// it is made, which is what lets an agent compose a pipeline instead of
/// running a call to discover the shape and then writing the real one.
///
/// `x-returns` is absent far more often than present, and deliberately so:
/// `crate::shapes::DECLARED` carries only shapes reproduced by calling the
/// builtin, so an absent key means "not established", never "returns nothing".
fn annotate_effects(mut defs: Vec<BuiltinDefinition>) -> Vec<BuiltinDefinition> {
    for d in &mut defs {
        let eff = crate::safety::effect_of(&d.name).as_str();
        if let JsonValue::Object(ref mut m) = d.json_schema {
            m.insert("x-effect".to_string(), json!(eff));
            // Whether `x-effect` was *decided* or merely *defaulted*.
            //
            // `effect_of` falls through to `pure`, so "pure" alone conflates a
            // builtin somebody read and found harmless with one nobody has
            // looked at — 694 of 1,301 are currently the latter. Policy is
            // right to treat them alike; an agent reading the ontology to
            // decide how much to trust the label is not, and had no way to
            // tell. Same convention as `x-returns`: this reports what is
            // established, never what is assumed.
            m.insert(
                "x-effect-declared".to_string(),
                json!(crate::safety::effect_is_declared(&d.name)),
            );
            // Where a field's *format* is surprising, an example goes with the
            // type. `ext:str` is true and still let a filter be written as
            // `f.ext == "rs"` when the value is `".rs"` — which returned an
            // empty set rather than an error.
            let examples = crate::shapes::field_examples(&d.name);
            if !examples.is_empty() {
                m.insert(
                    "x-returns-examples".to_string(),
                    json!(examples
                        .into_iter()
                        .map(|(f, e)| (f.to_string(), e.to_string()))
                        .collect::<std::collections::BTreeMap<_, _>>()),
                );
            }
            if let Some(shape) = crate::shapes::shape_of(&d.name) {
                m.insert("x-returns".to_string(), json!(shape));
            } else if let Some(shape) = crate::shapes::polymorphic_shape_of(&d.name) {
                // `T` is the element type of the first argument. Stated as a
                // relation rather than a concrete type because that is what is
                // true of `first`/`take`/`values`, and a concrete answer would
                // only describe whichever call happened to be measured.
                m.insert("x-returns".to_string(), json!(shape));
                m.insert(
                    "x-returns-var".to_string(),
                    json!("T = element type of the first argument"),
                );
            }
        }
    }
    defs
}

/// Get builtin definitions from documentation
fn get_builtin_definitions() -> Vec<BuiltinDefinition> {
    // Core builtins with full definitions
    let builtins = vec![
        // File System
        BuiltinDefinition {
            name: "ls".to_string(),
            description: "List directory contents".to_string(),
            category: "FileSystem".to_string(),
            signature: "ls(path?: String) -> Array<Record>".to_string(),
            parameters: vec![ParameterDefinition {
                name: "path".to_string(),
                param_type: "String".to_string(),
                description: "Directory path to list".to_string(),
                required: false,
                default: Some("\".\"".to_string()),
            }],
            return_type: "Array<Record>".to_string(),
            examples: vec![
                ExampleDefinition {
                    description: "List current directory".to_string(),
                    code: "ls()".to_string(),
                    result: Some("[{name: \"file.txt\", size: 1024, ...}, ...]".to_string()),
                },
                ExampleDefinition {
                    description: "List specific directory".to_string(),
                    code: "ls(\"/home\")".to_string(),
                    result: None,
                },
            ],
            aliases: Some(vec!["dir".to_string()]),
            json_schema: json!({
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Directory path"}
                }
            }),
        },
        BuiltinDefinition {
            name: "cat".to_string(),
            description: "Read file contents".to_string(),
            category: "FileSystem".to_string(),
            signature: "cat(path: String) -> String".to_string(),
            parameters: vec![ParameterDefinition {
                name: "path".to_string(),
                param_type: "String".to_string(),
                description: "File path to read".to_string(),
                required: true,
                default: None,
            }],
            return_type: "String".to_string(),
            examples: vec![ExampleDefinition {
                description: "Read a file".to_string(),
                code: "cat(\"README.md\")".to_string(),
                result: None,
            }],
            aliases: Some(vec!["read".to_string(), "read_text".to_string()]),
            json_schema: json!({
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path to read"}
                },
                "required": ["path"]
            }),
        },
        BuiltinDefinition {
            name: "pwd".to_string(),
            description: "Get current working directory".to_string(),
            category: "FileSystem".to_string(),
            signature: "pwd() -> String".to_string(),
            parameters: vec![],
            return_type: "String".to_string(),
            examples: vec![ExampleDefinition {
                description: "Get current directory".to_string(),
                code: "pwd()".to_string(),
                result: Some("\"/home/user\"".to_string()),
            }],
            aliases: None,
            json_schema: json!({"type": "object", "properties": {}}),
        },
        BuiltinDefinition {
            name: "cd".to_string(),
            description: "Change current directory".to_string(),
            category: "FileSystem".to_string(),
            signature: "cd(path: String) -> String".to_string(),
            parameters: vec![ParameterDefinition {
                name: "path".to_string(),
                param_type: "String".to_string(),
                description: "Directory to change to".to_string(),
                required: true,
                default: None,
            }],
            return_type: "String".to_string(),
            examples: vec![ExampleDefinition {
                description: "Change to home directory".to_string(),
                code: "cd(\"~\")".to_string(),
                result: None,
            }],
            aliases: None,
            json_schema: json!({
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Directory path"}
                },
                "required": ["path"]
            }),
        },
        // Functional
        BuiltinDefinition {
            name: "map".to_string(),
            description: "Transform each element of an array".to_string(),
            category: "Functional".to_string(),
            signature: "map(array: Array, fn: Lambda) -> Array".to_string(),
            parameters: vec![
                ParameterDefinition {
                    name: "array".to_string(),
                    param_type: "Array".to_string(),
                    description: "Input array (or piped input)".to_string(),
                    required: true,
                    default: None,
                },
                ParameterDefinition {
                    name: "fn".to_string(),
                    param_type: "Lambda".to_string(),
                    description: "Transformation function".to_string(),
                    required: true,
                    default: None,
                },
            ],
            return_type: "Array".to_string(),
            examples: vec![
                ExampleDefinition {
                    description: "Double each number".to_string(),
                    code: "[1, 2, 3] | map(fn(x) => x * 2)".to_string(),
                    result: Some("[2, 4, 6]".to_string()),
                },
                ExampleDefinition {
                    description: "Extract field from records".to_string(),
                    code: "ls() | map(fn(f) => f.name)".to_string(),
                    result: None,
                },
            ],
            aliases: None,
            json_schema: json!({
                "type": "object",
                "properties": {
                    "input": {"type": "array", "description": "Array to transform"},
                    "field": {"type": "string", "description": "Field to extract (shorthand for map)"}
                }
            }),
        },
        BuiltinDefinition {
            name: "where".to_string(),
            description: "Filter array elements by predicate".to_string(),
            category: "Functional".to_string(),
            signature: "where(array: Array, predicate: Lambda) -> Array".to_string(),
            parameters: vec![
                ParameterDefinition {
                    name: "array".to_string(),
                    param_type: "Array".to_string(),
                    description: "Input array".to_string(),
                    required: true,
                    default: None,
                },
                ParameterDefinition {
                    name: "predicate".to_string(),
                    param_type: "Lambda".to_string(),
                    description: "Filter function returning Bool".to_string(),
                    required: true,
                    default: None,
                },
            ],
            return_type: "Array".to_string(),
            examples: vec![ExampleDefinition {
                description: "Filter large files".to_string(),
                code: "ls() | where(fn(f) => f.size > 1000)".to_string(),
                result: None,
            }],
            aliases: Some(vec!["filter".to_string()]),
            json_schema: json!({
                "type": "object",
                "properties": {
                    "input": {"type": "array"},
                    "predicate": {"type": "string", "description": "Filter condition like 'size > 1000'"}
                }
            }),
        },
        BuiltinDefinition {
            name: "reduce".to_string(),
            description: "Reduce array to single value".to_string(),
            category: "Functional".to_string(),
            signature: "reduce(array: Array, fn: Lambda, init: Value) -> Value".to_string(),
            parameters: vec![
                ParameterDefinition {
                    name: "array".to_string(),
                    param_type: "Array".to_string(),
                    description: "Input array".to_string(),
                    required: true,
                    default: None,
                },
                ParameterDefinition {
                    name: "fn".to_string(),
                    param_type: "Lambda".to_string(),
                    description: "Reducer function (acc, item) => new_acc".to_string(),
                    required: true,
                    default: None,
                },
                ParameterDefinition {
                    name: "init".to_string(),
                    param_type: "Value".to_string(),
                    description: "Initial accumulator value".to_string(),
                    required: true,
                    default: None,
                },
            ],
            return_type: "Value".to_string(),
            examples: vec![ExampleDefinition {
                description: "Sum numbers".to_string(),
                code: "[1, 2, 3] | reduce(fn(a, b) => a + b, 0)".to_string(),
                result: Some("6".to_string()),
            }],
            aliases: Some(vec!["fold".to_string()]),
            json_schema: json!({
                "type": "object",
                "properties": {
                    "input": {"type": "array"},
                    "operation": {"type": "string", "enum": ["sum", "product", "concat", "custom"]},
                    "initial": {"description": "Initial value"}
                }
            }),
        },
        // AI
        BuiltinDefinition {
            name: "ai".to_string(),
            description: "Query AI model with a prompt".to_string(),
            category: "AI".to_string(),
            signature: "ai(prompt?: String, config?: Record) -> String".to_string(),
            parameters: vec![
                ParameterDefinition {
                    name: "prompt".to_string(),
                    param_type: "String".to_string(),
                    description: "The prompt to send to the AI; with piped input the piped string is the prompt".to_string(),
                    required: false,
                    default: None,
                },
                ParameterDefinition {
                    name: "config".to_string(),
                    param_type: "Record".to_string(),
                    description: "Optional images, audio or video arrays of file paths; a model field is accepted and not used".to_string(),
                    required: false,
                    default: None,
                },
            ],
            return_type: "String".to_string(),
            examples: vec![ExampleDefinition {
                description: "Ask AI a question".to_string(),
                code: "ai(\"Explain recursion\")".to_string(),
                result: None,
            }],
            // `ask` was listed here but never dispatched: calling it was
            // E_UNKNOWN_BUILTIN. tests/documented_builtin_count.rs now checks every
            // advertised alias against the dispatcher.
            aliases: None,
            json_schema: json!({
                "type": "object",
                "properties": {
                    "prompt": {"type": "string", "description": "AI prompt"},
                    "model": {"type": "string", "description": "Model to use"},
                    "temperature": {"type": "number", "minimum": 0, "maximum": 2}
                },
                "required": ["prompt"]
            }),
        },
        // Aggregation
        BuiltinDefinition {
            name: "sum".to_string(),
            description: "Sum numeric values in an array".to_string(),
            category: "Aggregation".to_string(),
            signature: "sum(array: Array<Number>) -> Number".to_string(),
            parameters: vec![ParameterDefinition {
                name: "array".to_string(),
                param_type: "Array<Number>".to_string(),
                description: "Array of numbers".to_string(),
                required: true,
                default: None,
            }],
            return_type: "Number".to_string(),
            examples: vec![ExampleDefinition {
                description: "Sum numbers".to_string(),
                code: "[1, 2, 3, 4, 5] | sum()".to_string(),
                result: Some("15".to_string()),
            }],
            aliases: None,
            json_schema: json!({
                "type": "object",
                "properties": {
                    "input": {"type": "array", "items": {"type": "number"}}
                }
            }),
        },
        BuiltinDefinition {
            name: "avg".to_string(),
            description: "Calculate average of numeric values".to_string(),
            category: "Aggregation".to_string(),
            signature: "avg(array: Array<Number>) -> Float".to_string(),
            parameters: vec![ParameterDefinition {
                name: "array".to_string(),
                param_type: "Array<Number>".to_string(),
                description: "Array of numbers".to_string(),
                required: true,
                default: None,
            }],
            return_type: "Float".to_string(),
            examples: vec![ExampleDefinition {
                description: "Average of numbers".to_string(),
                code: "[1, 2, 3, 4, 5] | avg()".to_string(),
                result: Some("3.0".to_string()),
            }],
            aliases: Some(vec!["mean".to_string()]),
            json_schema: json!({
                "type": "object",
                "properties": {
                    "input": {"type": "array", "items": {"type": "number"}}
                }
            }),
        },
        BuiltinDefinition {
            name: "len".to_string(),
            description: "Get length of array or string".to_string(),
            category: "Core".to_string(),
            signature: "len(value: Array | String) -> Int".to_string(),
            parameters: vec![ParameterDefinition {
                name: "value".to_string(),
                param_type: "Array | String".to_string(),
                description: "Array or string to measure".to_string(),
                required: true,
                default: None,
            }],
            return_type: "Int".to_string(),
            examples: vec![
                ExampleDefinition {
                    description: "Array length".to_string(),
                    code: "len([1, 2, 3])".to_string(),
                    result: Some("3".to_string()),
                },
                ExampleDefinition {
                    description: "String length".to_string(),
                    code: "len(\"hello\")".to_string(),
                    result: Some("5".to_string()),
                },
            ],
            aliases: Some(vec!["length".to_string(), "count".to_string()]),
            json_schema: json!({
                "type": "object",
                "properties": {
                    "input": {"description": "Array or string"}
                }
            }),
        },
        // String operations
        BuiltinDefinition {
            name: "split".to_string(),
            description: "Split string into array".to_string(),
            category: "String".to_string(),
            signature: "split(str: String, delimiter: String) -> Array<String>".to_string(),
            parameters: vec![
                ParameterDefinition {
                    name: "str".to_string(),
                    param_type: "String".to_string(),
                    description: "String to split".to_string(),
                    required: true,
                    default: None,
                },
                ParameterDefinition {
                    name: "delimiter".to_string(),
                    param_type: "String".to_string(),
                    description: "Delimiter to split on".to_string(),
                    required: true,
                    default: None,
                },
            ],
            return_type: "Array<String>".to_string(),
            examples: vec![ExampleDefinition {
                description: "Split CSV line".to_string(),
                code: "\"a,b,c\" | split(\",\")".to_string(),
                result: Some("[\"a\", \"b\", \"c\"]".to_string()),
            }],
            aliases: None,
            json_schema: json!({
                "type": "object",
                "properties": {
                    "input": {"type": "string"},
                    "delimiter": {"type": "string"}
                },
                "required": ["delimiter"]
            }),
        },
        BuiltinDefinition {
            name: "join".to_string(),
            description: "Join array elements into string".to_string(),
            category: "String".to_string(),
            signature: "join(array: Array, delimiter: String) -> String".to_string(),
            parameters: vec![
                ParameterDefinition {
                    name: "array".to_string(),
                    param_type: "Array".to_string(),
                    description: "Array to join".to_string(),
                    required: true,
                    default: None,
                },
                ParameterDefinition {
                    name: "delimiter".to_string(),
                    param_type: "String".to_string(),
                    description: "Delimiter between elements".to_string(),
                    required: true,
                    default: None,
                },
            ],
            return_type: "String".to_string(),
            examples: vec![ExampleDefinition {
                description: "Join with comma".to_string(),
                code: "[\"a\", \"b\", \"c\"] | join(\", \")".to_string(),
                result: Some("\"a, b, c\"".to_string()),
            }],
            aliases: None,
            json_schema: json!({
                "type": "object",
                "properties": {
                    "input": {"type": "array"},
                    "delimiter": {"type": "string"}
                },
                "required": ["delimiter"]
            }),
        },
        // HTTP
        BuiltinDefinition {
            name: "http_get".to_string(),
            description: "Make HTTP GET request".to_string(),
            category: "Network".to_string(),
            signature: "http_get(url: String, options?: Record) -> Value".to_string(),
            parameters: vec![
                ParameterDefinition {
                    name: "url".to_string(),
                    param_type: "String".to_string(),
                    description: "URL to request".to_string(),
                    required: true,
                    default: None,
                },
                ParameterDefinition {
                    name: "options".to_string(),
                    param_type: "Record".to_string(),
                    description: "Headers, timeout, etc.".to_string(),
                    required: false,
                    default: None,
                },
            ],
            return_type: "Value".to_string(),
            examples: vec![ExampleDefinition {
                description: "Fetch JSON API".to_string(),
                code: "http_get(\"https://api.example.com/data\")".to_string(),
                result: None,
            }],
            aliases: Some(vec!["fetch".to_string(), "get".to_string()]),
            json_schema: json!({
                "type": "object",
                "properties": {
                    "url": {"type": "string", "format": "uri"},
                    "headers": {"type": "object"}
                },
                "required": ["url"]
            }),
        },
        // MCP Tools
        BuiltinDefinition {
            name: "mcp_tools".to_string(),
            description: "List available MCP tools".to_string(),
            category: "MCP".to_string(),
            signature: "mcp_tools(options?: Record) -> Array<Record>".to_string(),
            parameters: vec![ParameterDefinition {
                name: "options".to_string(),
                param_type: "Record".to_string(),
                description: "Filter options like category".to_string(),
                required: false,
                default: None,
            }],
            return_type: "Array<Record>".to_string(),
            examples: vec![ExampleDefinition {
                description: "List all tools".to_string(),
                code: "mcp_tools()".to_string(),
                result: None,
            }],
            aliases: None,
            json_schema: json!({
                "type": "object",
                "properties": {
                    "category": {"type": "string"}
                }
            }),
        },
        BuiltinDefinition {
            name: "mcp_call".to_string(),
            description: "Execute an MCP tool".to_string(),
            category: "MCP".to_string(),
            signature: "mcp_call(tool: String, args: Record) -> Value".to_string(),
            parameters: vec![
                ParameterDefinition {
                    name: "tool".to_string(),
                    param_type: "String".to_string(),
                    description: "Tool name".to_string(),
                    required: true,
                    default: None,
                },
                ParameterDefinition {
                    name: "args".to_string(),
                    param_type: "Record".to_string(),
                    description: "Tool arguments".to_string(),
                    required: true,
                    default: None,
                },
            ],
            return_type: "Value".to_string(),
            examples: vec![ExampleDefinition {
                description: "Call git tool".to_string(),
                code: "mcp_call(\"git\", {command: \"status\"})".to_string(),
                result: None,
            }],
            aliases: None,
            json_schema: json!({
                "type": "object",
                "properties": {
                    "tool": {"type": "string"},
                    "args": {"type": "object"}
                },
                "required": ["tool", "args"]
            }),
        },
    ];

    // More builtins can be added here or discovered dynamically

    annotate_effects(builtins)
}

/// Categorize a builtin by name using comprehensive prefix and name matching
fn categorize_builtin(name: &str) -> String {
    match name {
        // AI & Agents
        n if n.starts_with("ai") || n.starts_with("agent") || n.starts_with("swarm") => {
            "AI".to_string()
        }
        // MCP Protocol
        n if n.starts_with("mcp_") => "MCP".to_string(),
        // A2A / A2UI / NANDA protocols
        n if n.starts_with("a2a_") => "A2A".to_string(),
        n if n.starts_with("a2ui_") => "A2UI".to_string(),
        n if n.starts_with("nanda_") => "NANDA".to_string(),
        // ML / Neural Nets / Evolution / RL
        n if n.starts_with("nn_") => "NeuralNet".to_string(),
        n if n.starts_with("evo_") => "Evolution".to_string(),
        n if n.starts_with("rl_") => "ReinforcementLearning".to_string(),
        n if n.starts_with("kg_")
            || n.starts_with("rag_")
            || n.starts_with("semantic_cache")
            || n.starts_with("fine_tune") =>
        {
            "Knowledge".to_string()
        }
        // Network & HTTP
        n if n.starts_with("http_") || n.starts_with("net_") || n == "http_get" || n == "fetch" => {
            "Network".to_string()
        }
        // System info
        n if n.starts_with("sys_") || n == "whoami" || n == "hostname" || n == "uptime" => {
            "System".to_string()
        }
        // Process management
        n if n.starts_with("proc_") || n.starts_with("ps") || n == "kill" || n == "killall" => {
            "Process".to_string()
        }
        // File system
        n if n.starts_with("file_")
            || n.starts_with("fs_")
            || [
                "ls",
                "cat",
                "pwd",
                "cd",
                "mkdir",
                "rm",
                "mv",
                "cp",
                "find",
                "head",
                "tail",
                "wc",
                "sort",
                "uniq",
                "grep",
                "touch",
                "chmod",
                "chown",
                "stat",
                "glob",
                "tree",
                "exists",
                "read_text",
                "write",
                "list",
                "dir",
            ]
            .contains(&n) =>
        {
            "FileSystem".to_string()
        }
        // Services & daemons
        n if n.starts_with("svc_") || n.starts_with("service_") || n.starts_with("systemctl") => {
            "Service".to_string()
        }
        // Cron/scheduling
        n if n.starts_with("cron_") || n.starts_with("at_") => "Scheduling".to_string(),
        // Archive/compression
        n if n.starts_with("archive_")
            || n.starts_with("zip")
            || n.starts_with("tar")
            || n.starts_with("gzip")
            || n.starts_with("bzip2")
            || n.starts_with("xz") =>
        {
            "Archive".to_string()
        }
        // User/permission management
        n if n.starts_with("user_") || n.starts_with("perm_") || n.starts_with("group_") => {
            "UserManagement".to_string()
        }
        // Package management
        n if n.starts_with("pkg_") => "Package".to_string(),
        // Hardware/device
        n if n.starts_with("hw_") || n.starts_with("device_") => "Hardware".to_string(),
        // GUI automation
        n if n.starts_with("gui_")
            || n.starts_with("screen")
            || n.starts_with("click")
            || n.starts_with("type_text") =>
        {
            "GUI".to_string()
        }
        // Web automation
        n if n.starts_with("web_") || n.starts_with("browser_") => "Web".to_string(),
        // Clipboard & input
        n if n.starts_with("clip_") || n.starts_with("input_") => "Input".to_string(),
        // Database
        n if n.starts_with("db_") || n.starts_with("sqlite_") || n.starts_with("sql_") => {
            "Database".to_string()
        }
        // Security & Crypto
        n if n.starts_with("crypto_")
            || n.starts_with("hash")
            || n.starts_with("encrypt")
            || n.starts_with("decrypt")
            || n.starts_with("sign")
            || n.starts_with("verify")
            || n.starts_with("ssl_") =>
        {
            "Crypto".to_string()
        }
        // Containers
        n if n.starts_with("docker_")
            || n.starts_with("podman_")
            || n.starts_with("container_")
            || n.starts_with("lxc_") =>
        {
            "Container".to_string()
        }
        // Kubernetes
        n if n.starts_with("k8s_") || n.starts_with("helm_") || n.starts_with("kubectl_") => {
            "Kubernetes".to_string()
        }
        // VM/Hypervisor
        n if n.starts_with("vm_")
            || n.starts_with("hyperv_")
            || n.starts_with("virsh_")
            || n.starts_with("qemu_")
            || n.starts_with("wsl_") =>
        {
            "Virtualization".to_string()
        }
        // Cloud/IaC
        n if n.starts_with("terraform_")
            || n.starts_with("ansible_")
            || n.starts_with("pulumi_")
            || n.starts_with("vagrant_")
            || n.starts_with("packer_")
            || n.starts_with("cloud_") =>
        {
            "Cloud".to_string()
        }
        // Remote access
        n if n.starts_with("ssh_")
            || n.starts_with("scp_")
            || n.starts_with("rsync_")
            || n.starts_with("rdp_") =>
        {
            "RemoteAccess".to_string()
        }
        // Security
        n if n.starts_with("firewall_")
            || n.starts_with("selinux_")
            || n.starts_with("apparmor_") =>
        {
            "Security".to_string()
        }
        // Monitoring
        n if n.starts_with("monitor_")
            || n.starts_with("perf_")
            || n.starts_with("netstat_")
            || n.starts_with("watch_") =>
        {
            "Monitoring".to_string()
        }
        // Enterprise
        n if n.starts_with("rbac_") => "RBAC".to_string(),
        n if n.starts_with("audit_") => "Audit".to_string(),
        n if n.starts_with("sso_") => "SSO".to_string(),
        // Cluster/distributed
        n if n.starts_with("cluster_") || n.starts_with("job_") => "Cluster".to_string(),
        // Cloud & Scale
        n if n.starts_with("repl_") => "REPL".to_string(),
        n if n.starts_with("workspace_") => "Workspace".to_string(),
        n if n.starts_with("marketplace_") => "Marketplace".to_string(),
        n if n.starts_with("telemetry_") => "Telemetry".to_string(),
        // Git/code/dev tools
        n if n.starts_with("git_") => "Git".to_string(),
        n if n.starts_with("code_") => "Code".to_string(),
        n if n.starts_with("project_") => "Project".to_string(),
        n if n.starts_with("search_") => "Search".to_string(),
        n if n.starts_with("test_") => "Testing".to_string(),
        n if n.starts_with("diag_") => "Diagnostics".to_string(),
        n if n.starts_with("refactor_") => "Refactoring".to_string(),
        n if n.starts_with("session_") => "Session".to_string(),
        n if n.starts_with("docs_") || n.starts_with("doc_") => "Documentation".to_string(),
        n if n.starts_with("devenv_") => "DevEnvironment".to_string(),
        n if n.starts_with("platform_") => "Platform".to_string(),
        // Math
        n if n.starts_with("math_")
            || [
                "sqrt", "pow", "abs", "ceil", "floor", "round", "sin", "cos", "tan", "log", "exp",
                "pi", "e",
            ]
            .contains(&n) =>
        {
            "Math".to_string()
        }
        // String
        n if n.starts_with("str_")
            || [
                "split",
                "join",
                "trim",
                "upper",
                "lower",
                "replace",
                "contains",
                "starts_with",
                "ends_with",
                "pad_left",
                "pad_right",
                "repeat",
                "reverse",
                "slice",
                "format",
                "to_upper",
                "to_lower",
                "capitalize",
                "char_at",
                "substr",
                "str_replace",
            ]
            .contains(&n) =>
        {
            "String".to_string()
        }
        // Array/collection
        n if n.starts_with("arr_")
            || [
                "range",
                "flatten",
                "zip",
                "unique",
                "sort_by",
                "group_by",
                "chunk",
                "window",
                "scan",
                "zip_with",
                "interleave",
                "partition",
            ]
            .contains(&n) =>
        {
            "Array".to_string()
        }
        // JSON
        n if n.starts_with("json_") || n == "to_json" || n == "from_json" => "JSON".to_string(),
        // Shell
        n if n.starts_with("shell_") || n == "exec" || n == "source" || n == "eval_bash" => {
            "Shell".to_string()
        }
        // Functional
        n if [
            "map", "where", "reduce", "filter", "each", "any", "all", "take", "first", "last",
            "keys", "values", "select", "reject", "flat_map", "fold",
        ]
        .contains(&n) =>
        {
            "Functional".to_string()
        }
        // Aggregation
        n if [
            "sum", "avg", "mean", "min", "max", "count", "product", "len",
        ]
        .contains(&n) =>
        {
            "Aggregation".to_string()
        }
        // Core utilities
        _ => "Core".to_string(),
    }
}

/// Generate a human-readable description from a builtin name
fn describe_builtin_name(name: &str) -> String {
    // Convert snake_case to "Title Case Description"
    let words: Vec<&str> = name.split('_').collect();
    if words.len() <= 1 {
        // Single word builtins
        return match name {
            "ls" => "List directory contents".to_string(),
            "cat" => "Read file contents".to_string(),
            "pwd" => "Print working directory".to_string(),
            "cd" => "Change directory".to_string(),
            "rm" => "Remove file or directory".to_string(),
            "mv" => "Move or rename file".to_string(),
            "cp" => "Copy file or directory".to_string(),
            "grep" => "Search for pattern in text".to_string(),
            "sort" => "Sort lines or array elements".to_string(),
            "uniq" => {
                "Remove adjacent duplicate elements (like Unix uniq); use unique for all duplicates"
                    .to_string()
            }
            "wc" => "Count lines, words, and bytes".to_string(),
            "head" => "Show first N lines".to_string(),
            "tail" => "Show last N lines".to_string(),
            "find" => "Find files by pattern".to_string(),
            "mkdir" => "Create directory".to_string(),
            "touch" => "Create empty file or update timestamp".to_string(),
            "echo" => "Print value to output".to_string(),
            "print" => "Print formatted output".to_string(),
            "help" => "Show available commands and help".to_string(),
            "clear" => "Clear terminal screen".to_string(),
            "map" => "Transform each element with a function".to_string(),
            "where" | "filter" => "Filter elements by predicate".to_string(),
            "reduce" => "Reduce array to single value".to_string(),
            "sum" => "Sum numeric values".to_string(),
            "avg" => "Calculate average of numeric values".to_string(),
            "len" => "Get length of array or string".to_string(),
            "keys" => "Get keys of a record".to_string(),
            "values" => "Get values of a record".to_string(),
            "first" => "Get first element".to_string(),
            "last" => "Get last element".to_string(),
            "any" => "Check if any element matches".to_string(),
            "all" => "Check if all elements match".to_string(),
            "take" => "Take first N elements".to_string(),
            "each" => "Execute function for each element".to_string(),
            "split" => "Split string by delimiter".to_string(),
            "join" => "Join array elements with delimiter".to_string(),
            "trim" => "Remove whitespace from string".to_string(),
            "upper" => "Convert string to uppercase".to_string(),
            "lower" => "Convert string to lowercase".to_string(),
            "contains" => "Check if string contains substring".to_string(),
            "replace" => "Replace occurrences in string".to_string(),
            "min" => "Get minimum value".to_string(),
            "max" => "Get maximum value".to_string(),
            "count" => "Count elements".to_string(),
            "flatten" => "Flatten nested arrays".to_string(),
            "range" => "Generate a range of numbers".to_string(),
            "unique" => "Remove duplicate elements".to_string(),
            "reverse" => "Reverse array or string".to_string(),
            "select" => "Select fields from records".to_string(),
            "reject" => "Remove fields from records".to_string(),
            "call" => "Call a named builtin dynamically".to_string(),
            "type" => "Get the type of a value".to_string(),
            "whoami" => "Get current username".to_string(),
            "hostname" => "Get system hostname".to_string(),
            "uptime" => "Get system uptime".to_string(),
            "kill" => "Terminate a process".to_string(),
            "exec" => "Execute external command".to_string(),
            "fetch" => "Fetch URL contents".to_string(),
            "glob" => "Match files by glob pattern".to_string(),
            _ => capitalize_first(name).to_string(),
        };
    }

    // Multi-word: use module prefix as context
    let prefix = words[0];
    let rest: Vec<&str> = words[1..].to_vec();
    let action = rest.join(" ");

    match prefix {
        "sys" => format!("System: {}", action),
        "proc" => format!("Process: {}", action),
        "net" => format!("Network: {}", action),
        "http" => format!("HTTP: {}", action),
        "file" => format!("File: {}", action),
        "fs" => format!("Filesystem: {}", action),
        "gui" => format!("GUI: {}", action),
        "web" => format!("Web: {}", action),
        "db" | "sqlite" => format!("Database: {}", action),
        "crypto" => format!("Crypto: {}", action),
        "svc" | "service" => format!("Service: {}", action),
        "cron" => format!("Schedule: {}", action),
        "archive" => format!("Archive: {}", action),
        "user" => format!("User: {}", action),
        "perm" => format!("Permission: {}", action),
        "pkg" => format!("Package: {}", action),
        "hw" => format!("Hardware: {}", action),
        "clip" => format!("Clipboard: {}", action),
        "input" => format!("Input: {}", action),
        "docker" => format!("Docker: {}", action),
        "podman" => format!("Podman: {}", action),
        "container" => format!("Container: {}", action),
        "k8s" | "kubectl" => format!("Kubernetes: {}", action),
        "helm" => format!("Helm: {}", action),
        "vm" => format!("VM: {}", action),
        "hyperv" => format!("Hyper-V: {}", action),
        "virsh" => format!("Libvirt: {}", action),
        "qemu" => format!("QEMU: {}", action),
        "wsl" => format!("WSL: {}", action),
        "lxc" => format!("LXC: {}", action),
        "terraform" => format!("Terraform: {}", action),
        "ansible" => format!("Ansible: {}", action),
        "pulumi" => format!("Pulumi: {}", action),
        "vagrant" => format!("Vagrant: {}", action),
        "packer" => format!("Packer: {}", action),
        "ssh" => format!("SSH: {}", action),
        "scp" => format!("SCP: {}", action),
        "rsync" => format!("Rsync: {}", action),
        "rdp" => format!("RDP: {}", action),
        "firewall" => format!("Firewall: {}", action),
        "selinux" => format!("SELinux: {}", action),
        "apparmor" => format!("AppArmor: {}", action),
        "ssl" => format!("SSL/TLS: {}", action),
        "monitor" => format!("Monitor: {}", action),
        "perf" => format!("Performance: {}", action),
        "git" => format!("Git: {}", action),
        "code" => format!("Code: {}", action),
        "project" => format!("Project: {}", action),
        "search" => format!("Search: {}", action),
        "test" => format!("Test: {}", action),
        "diag" => format!("Diagnostics: {}", action),
        "refactor" => format!("Refactor: {}", action),
        "session" => format!("Session: {}", action),
        "docs" | "doc" => format!("Documentation: {}", action),
        "devenv" => format!("Dev environment: {}", action),
        "platform" => format!("Platform: {}", action),
        "math" => format!("Math: {}", action),
        "str" => format!("String: {}", action),
        "arr" => format!("Array: {}", action),
        "json" => format!("JSON: {}", action),
        "shell" => format!("Shell: {}", action),
        "mcp" => format!("MCP: {}", action),
        "a2a" => format!("Agent-to-Agent: {}", action),
        "a2ui" => format!("Agent-to-UI: {}", action),
        "nanda" => format!("NANDA: {}", action),
        "nn" => format!("Neural net: {}", action),
        "evo" => format!("Evolution: {}", action),
        "rl" => format!("Reinforcement learning: {}", action),
        "rbac" => format!("RBAC: {}", action),
        "audit" => format!("Audit: {}", action),
        "sso" => format!("SSO: {}", action),
        "cluster" => format!("Cluster: {}", action),
        "cloud" => format!("Cloud: {}", action),
        "repl" => format!("REPL: {}", action),
        "workspace" => format!("Workspace: {}", action),
        "marketplace" => format!("Marketplace: {}", action),
        "telemetry" => format!("Telemetry: {}", action),
        _ => {
            let full = words.join(" ");
            capitalize_first(&full)
        }
    }
}

/// Capitalize the first letter of a string
fn capitalize_first(s: &str) -> String {
    let mut chars = s.chars();
    match chars.next() {
        None => String::new(),
        Some(c) => c.to_uppercase().to_string() + chars.as_str(),
    }
}

/// Infer a basic return type from the builtin name
fn infer_return_type(name: &str) -> String {
    match name {
        // Known return types for common builtins
        n if n == "ls" || n == "find" || n == "glob" || n == "tree" => "Array<Record>".to_string(),
        n if n == "cat"
            || n == "pwd"
            || n == "whoami"
            || n == "hostname"
            || n == "echo"
            || n == "print" =>
        {
            "String".to_string()
        }
        n if n == "len" || n == "count" || n == "wc" => "Int".to_string(),
        n if n == "sum" || n == "avg" || n == "min" || n == "max" || n == "product" => {
            "Number".to_string()
        }
        n if n == "any" || n == "all" || n == "exists" || n == "contains" => "Bool".to_string(),
        n if n == "first" || n == "last" || n == "reduce" => "Value".to_string(),
        n if n == "map"
            || n == "where"
            || n == "filter"
            || n == "take"
            || n == "sort"
            || n == "uniq"
            || n == "unique"
            || n == "flatten"
            || n == "range"
            || n == "reverse"
            || n == "select"
            || n == "reject"
            || n == "keys"
            || n == "values"
            || n == "each" =>
        {
            "Array".to_string()
        }
        n if n == "split" => "Array<String>".to_string(),
        n if n == "join"
            || n == "trim"
            || n == "upper"
            || n == "lower"
            || n == "replace"
            || n == "format" =>
        {
            "String".to_string()
        }
        // Module-prefix patterns
        n if n.ends_with("_list")
            || n.ends_with("_all")
            || n.ends_with("_search")
            || n.ends_with("_query") =>
        {
            "Array<Record>".to_string()
        }
        n if n.ends_with("_info")
            || n.ends_with("_status")
            || n.ends_with("_config")
            || n.ends_with("_stats")
            || n.ends_with("_details") =>
        {
            "Record".to_string()
        }
        n if n.ends_with("_count")
            || n.ends_with("_size")
            || n.ends_with("_pid")
            || n.ends_with("_port") =>
        {
            "Int".to_string()
        }
        n if n.ends_with("_exists")
            || n.ends_with("_check")
            || n.ends_with("_validate")
            || n.ends_with("_enabled")
            || n.ends_with("_available") =>
        {
            "Bool".to_string()
        }
        n if n.ends_with("_read")
            || n.ends_with("_get")
            || n.ends_with("_name")
            || n.ends_with("_version")
            || n.ends_with("_path")
            || n.ends_with("_hostname") =>
        {
            "String".to_string()
        }
        // Prefix patterns for structured output
        n if n.starts_with("sys_") || n.starts_with("hw_") || n.starts_with("platform_") => {
            "Record".to_string()
        }
        n if n.starts_with("proc_") && n.ends_with("s") => "Array<Record>".to_string(),
        n if n.starts_with("net_") => "Record".to_string(),
        _ => "Value".to_string(),
    }
}

/// The doc of the `subject` parameter a subject-taking builtin advertises.
const SUBJECT_PARAM_DOC: &str =
    "the value the builtin acts on; in a pipeline it is the piped value";

/// Subject-taking builtins with no parameter that run with no subject at all.
/// Every other such builtin refuses a missing subject in its body (`print`,
/// `inspect`, `tokens`, `diagnose`, `sso_validate`, the `a2ui_*` builtins and
/// the rest checked), so an agent must supply one and the tool schema requires
/// it. Empty until a builtin is found that runs with nothing piped.
const SUBJECT_OPTIONAL_WITHOUT_PARAMS: &[&str] = &[];

/// Whether a tool schema requires the subject. A direct call has no pipeline,
/// so the subject is the first argument: it is required when the builtin is
/// pipeline-only, when a parameter is required (`pow(3)` reads its argument as
/// the subject), or when the builtin has no parameter to take the value in
/// (`sqrt(16)`), unless it is listed as running without one.
fn subject_required_in_tools(sig: &crate::signature::Signature) -> bool {
    // `sql("select 1 as x")` is a lone query: its one String is
    // `source_or_query`, not a subject, so the subject is never required.
    if matches!(sig.name, "sql" | "sql_value") {
        return false;
    }
    sig.subject_required
        || sig.params.iter().any(|p| p.required)
        || (sig.params.is_empty() && !SUBJECT_OPTIONAL_WITHOUT_PARAMS.contains(&sig.name))
}

/// Whether a parameter's doc says it takes the pipeline input when omitted
/// (`file_write`'s `content`). Such a parameter is required in a tool schema.
fn defaults_to_pipeline(p: &crate::signature::Param) -> bool {
    p.doc.contains("omitted uses pipeline input")
}

/// Build a catalogue entry from a declared signature, so the description an
/// agent reads and the check the dispatcher runs come from one place.
fn definition_from_signature(
    sig: &crate::signature::Signature,
    primary_name: &str,
    idx: &usize,
    index_to_names: &HashMap<usize, Vec<String>>,
) -> BuiltinDefinition {
    let aliases: Vec<String> = index_to_names
        .get(idx)
        .map(|names| {
            names
                .iter()
                .filter(|n| n.as_str() != primary_name)
                .cloned()
                .collect()
        })
        .unwrap_or_default();

    // A builtin with a declared subject takes it first. In a pipeline it is the
    // piped value. It is required when the builtin is pipeline-only
    // (`subject_required`), and also when it has a required parameter: a direct
    // call's first argument is read as the subject, so `pow(3)` cannot give the
    // exponent without one. Without it here an agent following the schema had
    // no parameter to put the string in.
    let subject_param = sig.subject.map(|ty| {
        let required = subject_required_in_tools(sig);
        (ty.as_str(), required, SUBJECT_PARAM_DOC)
    });
    // The schema lists `subject` first, then the parameters in declaration
    // order. That is the order the `properties` object keeps, because serde_json
    // is built with `preserve_order`; without it the map would sort them.
    let properties: serde_json::Map<String, JsonValue> = subject_param
        .iter()
        .map(|(ty, _, doc)| {
            (
                "subject".to_string(),
                json!({ "type": ty, "description": doc }),
            )
        })
        .chain(sig.params.iter().map(|p| {
            (
                p.name.to_string(),
                json!({ "type": p.ty.as_str(), "description": p.doc }),
            )
        }))
        .collect();
    // The tool schema's required list. A direct call has no pipeline, so a
    // parameter that takes the pipeline input when omitted is required here.
    let required: Vec<&str> = subject_param
        .iter()
        .filter(|(_, required, _)| *required)
        .map(|_| "subject")
        .chain(
            sig.params
                .iter()
                .filter(|p| p.required || defaults_to_pipeline(p))
                .map(|p| p.name),
        )
        .collect();

    BuiltinDefinition {
        name: primary_name.to_string(),
        description: sig.doc.to_string(),
        // A declared category wins. `categorize_builtin` works from the name,
        // which filed the array `zip` under Archive and `uniq` under
        // FileSystem -- and a category is how an agent browses, so a wrong one
        // hides the builtin from the list it belongs in.
        category: sig
            .category
            .map(str::to_string)
            .unwrap_or_else(|| categorize_builtin(primary_name)),
        signature: sig.render(),
        parameters: subject_param
            .iter()
            .map(|(ty, required, doc)| ParameterDefinition {
                name: "subject".to_string(),
                param_type: ty.to_string(),
                description: doc.to_string(),
                required: *required,
                default: None,
            })
            .chain(sig.params.iter().map(|p| ParameterDefinition {
                name: p.name.to_string(),
                param_type: p.ty.as_str().to_string(),
                description: p.doc.to_string(),
                required: p.required,
                default: None,
            }))
            .collect(),
        return_type: sig.returns.to_string(),
        examples: sig
            .examples
            .iter()
            .map(|(code, result)| ExampleDefinition {
                description: String::new(),
                code: (*code).to_string(),
                result: Some((*result).to_string()),
            })
            .collect(),
        aliases: if aliases.is_empty() {
            None
        } else {
            Some(aliases)
        },
        json_schema: json!({
            "type": "object",
            "properties": properties,
            "required": required,
        }),
    }
}

/// Generate all builtin definitions dynamically from BUILTIN_LOOKUP.
/// Enriched hand-coded definitions for core builtins are preserved;
/// all remaining builtins get auto-generated schemas.
fn get_all_builtin_definitions() -> Vec<BuiltinDefinition> {
    // Build reverse lookup: index → Vec<name> to detect aliases
    let mut index_to_names: HashMap<usize, Vec<String>> = HashMap::new();
    for (name, &idx) in BUILTIN_LOOKUP.iter() {
        index_to_names
            .entry(idx)
            .or_default()
            .push(name.to_string());
    }

    // Start with hand-coded enriched definitions — except where the builtin has
    // a declared signature, which outranks them. Two independent descriptions of
    // one builtin is the defect this work exists to remove, and the hand-written
    // entries are where it bit hardest: `map`'s said `map(array, fn) -> Array`
    // while the dispatcher checked something else. The declaration is what the
    // dispatcher enforces, so the catalogue must say what the declaration says.
    let enriched: Vec<BuiltinDefinition> = get_builtin_definitions()
        .into_iter()
        .map(|d| {
            match (
                crate::signature::signature_of(&d.name),
                BUILTIN_LOOKUP.get(d.name.as_str()),
            ) {
                (Some(sig), Some(idx)) => {
                    definition_from_signature(sig, &d.name, idx, &index_to_names)
                }
                _ => d,
            }
        })
        .collect();
    let enriched_names: std::collections::HashSet<String> =
        enriched.iter().map(|b| b.name.clone()).collect();

    // Determine the "canonical" name for each index (shortest non-alias name)
    // A hyphenated spelling is never the canonical one: `to-csv` and `to_csv`
    // share an index and are the same length, and `-` sorts before `_`, so
    // without this the hyphen was chosen and the underscore lost.
    let mut canonical: HashMap<usize, String> = HashMap::new();
    for (idx, names) in &index_to_names {
        let mut sorted = names.clone();
        sorted.sort_by(|a, b| {
            a.contains('-')
                .cmp(&b.contains('-'))
                .then(a.len().cmp(&b.len()))
                .then(a.cmp(b))
        });
        canonical.insert(*idx, sorted[0].clone());
    }

    // Build final list
    let mut all = enriched;
    let mut seen_indices: std::collections::HashSet<usize> = std::collections::HashSet::new();

    // Mark enriched indices as seen
    for name in &enriched_names {
        if let Some(&idx) = BUILTIN_LOOKUP.get(name.as_str()) {
            seen_indices.insert(idx);
        }
    }

    // Generate definitions for all remaining builtins
    for (idx, primary_name) in &canonical {
        if seen_indices.contains(idx) {
            continue;
        }
        seen_indices.insert(*idx);

        let category = categorize_builtin(primary_name);
        // A declared signature is the truth. Everything below it is inference
        // from the builtin's *name* -- `describe_builtin_name` splits on `_`
        // and title-cases the rest, so `db_json_to_sqlite` becomes "Database:
        // json to sqlite" with no parameters, and `max` became "Get maximum
        // value, max() -> Number" while refusing the array that implies. An
        // agent asking what a function does was being told what its name looks
        // like. See `crate::signature`.
        if let Some(sig) = crate::signature::signature_of(primary_name) {
            all.push(definition_from_signature(
                sig,
                primary_name,
                idx,
                &index_to_names,
            ));
            continue;
        }

        let description = describe_builtin_name(primary_name);
        let return_type = infer_return_type(primary_name);

        // Collect aliases (other names mapping to same index)
        let aliases: Vec<String> = index_to_names
            .get(idx)
            .map(|names| {
                names
                    .iter()
                    .filter(|n| *n != primary_name)
                    .cloned()
                    .collect()
            })
            .unwrap_or_default();

        let signature = format!("{}() -> {}", primary_name, return_type);

        all.push(BuiltinDefinition {
            name: primary_name.clone(),
            description,
            category,
            signature,
            parameters: vec![],
            return_type,
            examples: vec![],
            aliases: if aliases.is_empty() {
                None
            } else {
                Some(aliases)
            },
            json_schema: json!({
                "type": "object",
                "properties": {}
            }),
        });
    }

    // The dispatcher has two halves and this loop only walked one. Everything
    // in `FALLBACK_BUILTINS` -- `from_json`, `to_json`, `group_by`, `select`,
    // `columns`, `describe` and ~200 more -- dispatches perfectly and was
    // absent from the catalogue entirely, so an agent asking the shell what it
    // can call was told a subset and given no hint that it was a subset. Four
    // of the builtins our own E1 corpus needs were in that gap.
    //
    // Fallback arms share an implementation the way `BUILTIN_LOOKUP` aliases
    // share an index, so they are grouped by callee to get the same
    // canonical-name-plus-aliases treatment.
    let seen_names: std::collections::HashSet<String> =
        all.iter().map(|d| d.name.clone()).collect();
    let mut by_callee: HashMap<&str, Vec<&str>> = HashMap::new();
    for (name, callee) in crate::builtins::FALLBACK_BUILTINS {
        // `BUILTIN_LOOKUP` wins where both define a name; its entry is already
        // in `all` and is the one dispatch actually reaches.
        if BUILTIN_LOOKUP.contains_key(*name) {
            continue;
        }
        by_callee.entry(callee).or_default().push(name);
    }
    for (_, mut names) in by_callee {
        names.sort_by(|a, b| {
            a.contains('-')
                .cmp(&b.contains('-'))
                .then(a.len().cmp(&b.len()))
                .then(a.cmp(b))
        });
        let primary = names[0];
        if seen_names.contains(primary) {
            continue;
        }
        let aliases: Vec<String> = names[1..].iter().map(|n| (*n).to_string()).collect();
        if let Some(sig) = crate::signature::signature_of(primary) {
            let mut def = definition_from_signature(sig, primary, &usize::MAX, &HashMap::new());
            def.aliases = (!aliases.is_empty()).then_some(aliases);
            all.push(def);
            continue;
        }
        let return_type = infer_return_type(primary);
        all.push(BuiltinDefinition {
            name: primary.to_string(),
            description: describe_builtin_name(primary),
            category: categorize_builtin(primary),
            signature: format!("{primary}() -> {return_type}"),
            parameters: vec![],
            return_type,
            examples: vec![],
            aliases: (!aliases.is_empty()).then_some(aliases),
            json_schema: json!({ "type": "object", "properties": {} }),
        });
    }

    // The parser reads a name as an identifier, so a hyphenated name such as
    // `to-csv` cannot be written in a program. The catalogue advertises the
    // underscore form only. A hyphenated spelling stays callable wherever it
    // already was, but is not listed, here or as an alias.
    // A declared PowerShell name with no underscore form (`measure-object`,
    // `foreach-object`, `select-object`) is not callable from ish syntax, so
    // its declaration renders a call form nobody can write. Its entry is
    // dropped where the catalogue is built, whatever name the entry took.
    all.retain(|d| {
        let call_head = d.signature.split('(').next().unwrap_or("");
        !d.name.contains('-') && !call_head.contains('-')
    });
    for d in &mut all {
        if let Some(aliases) = d.aliases.as_mut() {
            aliases.retain(|a| !a.contains('-'));
        }
        if d.aliases.as_ref().is_some_and(|a| a.is_empty()) {
            d.aliases = None;
        }
    }

    // Sort by category then name for consistent output
    all.sort_by(|a, b| a.category.cmp(&b.category).then(a.name.cmp(&b.name)));
    // The calling schema is JSON Schema, whatever spelling the entry above
    // gave a type; the parameter list keeps IronShell's own type names.
    for d in &mut all {
        normalize_schema_types(&mut d.json_schema);
    }
    annotate_effects(all)
}

/// The effect class an annotated definition carries (from its `json_schema`).
fn def_effect(d: &BuiltinDefinition) -> &str {
    d.json_schema
        .get("x-effect")
        .and_then(|v| v.as_str())
        .unwrap_or("pure")
}

/// Whether [`def_effect`]'s answer was decided or defaulted.
///
/// `pure` is both a classification and the fall-through, so on its own it
/// cannot tell an agent whether anyone has looked. Today no builtin is
/// explicitly classified `Pure`, which makes every `pure` a default -- see
/// `tests/effect_provenance.rs`, which pins that and will fail the moment it
/// stops being true.
fn def_effect_declared(d: &BuiltinDefinition) -> bool {
    d.json_schema
        .get("x-effect-declared")
        .and_then(|v| v.as_bool())
        .unwrap_or(false)
}

/// Progressive ontology disclosure (§5.4): a compact root manifest — one entry
/// per category with builtin count, the effect classes present, and a few
/// sample names — plus the effect legend and a usage hint. Agents load this
/// cheaply, then `ontology_describe` only the slice they need, instead of
/// carrying the full multi-thousand-token ontology.
pub fn ontology_manifest_json() -> JsonValue {
    use std::collections::{BTreeMap, BTreeSet};
    let defs = get_all_builtin_definitions();
    // Keep the root deliberately tiny: per category, just the count and the
    // effect classes present. Names/signatures live in ontology_describe so the
    // root stays a cheap index (the standing-context win).
    let mut by_cat: BTreeMap<String, (usize, BTreeSet<String>)> = BTreeMap::new();
    for d in &defs {
        let entry = by_cat
            .entry(d.category.clone())
            .or_insert((0, BTreeSet::new()));
        entry.0 += 1;
        entry.1.insert(def_effect(d).to_string());
    }
    let categories: Vec<JsonValue> = by_cat
        .into_iter()
        .map(|(cat, (count, effects))| {
            json!({
                "category": cat,
                "builtins": count,
                "effects": effects.into_iter().collect::<Vec<_>>(),
            })
        })
        .collect();
    // The module namespaces an agent writes as `file.read(...)`. Taken from the
    // table the evaluator registers, so each name here resolves; the member
    // names are left out to keep the root small (the schema module filter has them).
    // Space-joined, not an array: the root has a token budget (tests/output_economy.rs).
    let mut modules: Vec<&'static str> = crate::modules::all_modules()
        .into_iter()
        .map(|(name, _)| name)
        .collect();
    modules.sort_unstable();
    modules.dedup();
    json!({
        "ontology": "manifest",
        "total_builtins": defs.len(),
        "categories": categories,
        "modules": modules.join(" "),
        "effect_legend": [
            "pure", "read_local", "write_local", "destructive",
            "process", "network", "exec", "privileged"
        ],
        "hint": "ontology_describe(\"<category>\") lists a category's builtins; ontology_describe(\"<builtin>\") returns full detail; modules lists namespaces (call as <module>.<member>); `ish agent schema --module <name>` lists a module's members"
    })
}

/// Flat tool specs for every builtin — `{name, description, signature, effect}` —
/// for exposing builtins as MCP tools annotated with their safety effect class.
pub fn builtin_tool_specs() -> Vec<JsonValue> {
    get_all_builtin_definitions()
        .iter()
        .map(|d| {
            json!({
                "name": d.name,
                "description": d.description,
                "signature": d.signature,
                "effect": def_effect(d),
                // Absent unless proven — see `crate::shapes`. A polymorphic
                // shape is given in terms of `T`, the first argument's element
                // type.
                "returns": crate::shapes::shape_of(&d.name)
                    .or_else(|| crate::shapes::polymorphic_shape_of(&d.name)),
            })
        })
        .collect()
}

/// Expand one slice of the ontology: a builtin name → full definition (with
/// effect, params, examples); a category name → its builtins (name, signature,
/// effect). Case-insensitive. Returns an `{error, hint}` record when unknown.
/// The `import` statement, as the schema's syntax section and
/// `ontology_describe("import")` state it.
const IMPORT_SYNTAX: &str =
    "import \"lib/x.ish\"  (the path is relative to the importing file; only pub names are visible)";

/// The category names the catalogue lists, sorted and without repeats.
pub fn ontology_category_names() -> Vec<String> {
    let mut names: Vec<String> = get_all_builtin_definitions()
        .into_iter()
        .map(|d| d.category)
        .collect();
    names.sort();
    names.dedup();
    names
}

pub fn ontology_describe_json(query: &str) -> JsonValue {
    let defs = get_all_builtin_definitions();
    // `file.replace` names the builtin that the call runs (`file_replace`), so
    // the dotted spelling describes that same entry.
    let resolved = query
        .trim()
        .split_once('.')
        .and_then(|(m, f)| crate::modules::builtin_for_member(m, f));
    // A hyphenated spelling names the underscore builtin: `to-csv` is `to_csv`.
    // Categories have no hyphens, so the rewrite cannot shadow one.
    let underscored = query.trim().replace('-', "_");
    let q = resolved.as_deref().unwrap_or(&underscored);

    // An exact, case-sensitive category name wins. Categories are capitalised
    // ("Platform") and builtins are not ("platform"), but the builtin lookup
    // below is case-insensitive -- so four categories were shadowed by a
    // builtin of the same name. ontology_manifest() advertised 69 Platform,
    // 30 AI, 10 Service and 3 Cluster builtins, and ontology_describe() on
    // each of those categories returned the single builtin `platform`, `ai`,
    // `service` or `cluster` instead. 112 builtins were listed as existing and
    // could not be enumerated. `ontology_describe("platform")` still returns
    // the builtin.
    let exact_cat: Vec<JsonValue> = defs
        .iter()
        .filter(|d| d.category == q)
        .map(|d| json!({ "name": d.name, "signature": d.signature, "effect": def_effect(d) }))
        .collect();
    if !exact_cat.is_empty() {
        return json!({ "category": q, "builtins": exact_cat });
    }

    // `import` is a statement, not a builtin, so the catalogue has no entry for
    // it. It is answered from the same text the schema's syntax section gives.
    if q == "import" {
        return json!({
            "syntax": "import",
            "form": IMPORT_SYNTAX,
            "description": "Loads another IronShell file. The path is resolved relative to the importing file, not the working directory. Only names marked `pub` in that file are visible to the importer.",
        });
    }

    // Resolve by alias as well as by canonical name. Every entry already
    // *lists* its aliases, so refusing to look one up was the catalogue
    // declining to answer a question it had the answer to: `mean` is an alias
    // of `avg`, it is the spelling our own benchmark corpus uses, and
    // `ontology_describe("mean")` said "not a known builtin".
    let found = defs
        .iter()
        .find(|d| d.name.eq_ignore_ascii_case(q))
        .or_else(|| {
            defs.iter().find(|d| {
                d.aliases
                    .as_ref()
                    .is_some_and(|a| a.iter().any(|x| x.eq_ignore_ascii_case(q)))
            })
        })
        // A bare session verb names the builtin its `session.` member runs:
        // `redo` is `session.redo`, which is `session_redo`. Only after every
        // builtin and category of that name has been looked for.
        .or_else(|| {
            let member = crate::modules::builtin_for_member("session", q)?;
            defs.iter().find(|d| d.name == member)
        });
    if let Some(d) = found {
        // The subject is the first parameter of a subject-taking builtin, and
        // in a pipeline it is the piped value. Marked so a reader knows it may
        // be piped rather than written out.
        let has_subject = crate::signature::signature_of(&d.name)
            .is_some_and(|s| s.subject.is_some() && !s.params.iter().any(|p| p.name == "subject"));
        let params: Vec<JsonValue> = d
            .parameters
            .iter()
            .enumerate()
            .map(|(i, p)| {
                let piped = has_subject && i == 0 && p.name == "subject";
                let mut param = json!({
                    "name": p.name,
                    "type": p.param_type,
                    "required": p.required,
                    "description": p.description,
                });
                if piped {
                    param["subject"] = json!(true);
                }
                param
            })
            .collect();
        let examples: Vec<JsonValue> = d
            .examples
            .iter()
            .map(|e| {
                let mut ex = json!({ "code": e.code });
                if let Some(result) = &e.result {
                    ex["result"] = json!(result);
                }
                if !e.description.is_empty() {
                    ex["description"] = json!(e.description);
                }
                ex
            })
            .collect();
        let mut detail = json!({
            "builtin": d.name,
            "category": d.category,
            "signature": d.signature,
            "effect": def_effect(d),
            "return_type": d.return_type,
            "description": d.description,
            "parameters": params,
            "examples": examples,
            "aliases": d.aliases,
            // The same schema the `describe` action returns for this builtin:
            // both read the one catalogue entry, so the tool schema and the
            // description cannot disagree.
            "json_schema": d.json_schema.clone(),
        });
        // An alias is documented as one: when the signature names a different
        // builtin, that is the one the call runs (`delete_lines` runs
        // `file_delete_lines`).
        if let Some(implemented_by) = d.signature.split('(').next().map(str::trim) {
            if !implemented_by.is_empty() && implemented_by != d.name {
                detail["implemented_by"] = json!(implemented_by);
            }
        }
        // Only when the effect is the *fall-through*, which is the only
        // direction an agent can act on: it says "this label is weaker than it
        // looks -- nobody classified this builtin".
        //
        // Measured, after guessing wrong about where the cost was: emitting it
        // unconditionally on every builtin added 10,312 tokens (+9.2%) to the
        // full-detail dump. Emitting only the exception costs a fraction of
        // that, and absence carries the ordinary meaning -- the effect came
        // from a rule.
        if !def_effect_declared(d) {
            if let Some(obj) = detail.as_object_mut() {
                obj.insert("effect_declared".to_string(), json!(false));
            }
        }
        return detail;
    }

    let in_cat: Vec<JsonValue> = defs
        .iter()
        .filter(|d| d.category.eq_ignore_ascii_case(q))
        .map(|d| json!({ "name": d.name, "signature": d.signature, "effect": def_effect(d) }))
        .collect();
    if !in_cat.is_empty() {
        return json!({ "category": q, "builtins": in_cat });
    }

    // Name the nearest real builtins, so a typo is one step from the answer
    // whichever surface (builtin, MCP, Agent API) asked.
    let near = crate::builtins::did_you_mean(q);
    let error = if near.is_empty() {
        format!("'{q}' is not a known category or builtin")
    } else {
        format!(
            "'{q}' is not a known category or builtin; did you mean {}?",
            near.join(", ")
        )
    };
    json!({
        "error": error,
        "hint": "call ontology_manifest() to see categories",
        "did_you_mean": near,
    })
}

/// Module descriptions for the ontology
fn module_description(name: &str) -> &str {
    match name {
        "sys" => "System information — hostname, OS, CPU, memory, uptime",
        "proc" => "Process management — list, kill, monitor processes",
        "fs" => "Filesystem operations — mount, disk, symlink",
        "file" => "File I/O — read, write, append, copy, move, edit",
        "net" => "Network — ping, DNS, interfaces, connections",
        "http" => "HTTP client — get, post, put, delete, request",
        "gui" => "GUI automation — screenshot, click, type, window control",
        "web" => "Web automation — browser control, scraping",
        "crypto" => "Cryptography — hash, encrypt, decrypt, sign, UUID",
        "db" => "Database — SQLite queries, connections",
        "svc" => "Service management — start, stop, status, enable",
        "cron" => "Scheduling — cron jobs, scheduled tasks",
        "archive" => "Compression — zip, tar, gzip, extract",
        "user" => "User management — create, delete, modify users",
        "perm" => "Permissions — chmod, chown, ACLs",
        "pkg" => "Package management — install, remove, search, update",
        "hw" => "Hardware info — CPU, GPU, disk, memory, battery",
        "clip" => "Clipboard — copy, paste, history",
        "input" => "User input — prompt, confirm, select",
        "ai" => "AI queries — multi-provider LLM completion",
        "agent" => "Agent framework — autonomous agents with tools",
        "math" => "Mathematics — sqrt, pow, trig, constants",
        "str" => "String manipulation — upper, lower, split, join, replace",
        "arr" => "Array operations — range, flatten, unique, chunk",
        "json" => "JSON — parse, stringify, query, patch",
        "mcp" => "Model Context Protocol — tool discovery and invocation",
        "shell" => "Shell operations — exec, source, environment",
        "a2ui" => "Agent-to-UI — notifications, progress, confirmations",
        "a2a" => "Agent-to-Agent — inter-agent messaging",
        "nanda" => "NANDA consensus protocol",
        "git" => "Git operations — status, commit, push, pull, branch",
        "code" => "Code intelligence — analysis, completion, generation",
        "project" => "Project management — scaffold, dependencies",
        "search" => "Code search — find symbols, references, definitions",
        "test" => "Testing — run, discover, analyze test results",
        "diag" => "Diagnostics — linting, type checking, profiling",
        "refactor" => "Refactoring — rename, extract, inline, organize",
        "session" => "Session management — save, restore, history",
        "docs" => "Documentation — generate, browse, search docs",
        "devenv" => "Dev environment — setup, tools, configuration",
        "platform" => "Platform detection — OS, arch, features, libraries",
        "rbac" => "Role-based access control — roles, permissions, grants",
        "audit" => "Audit logging — log, query, export events",
        "sso" => "Single sign-on — OIDC/SAML authentication",
        "cluster" => "Distributed computing — cluster, job scheduling",
        "nn" => "Neural networks — create, train, predict",
        "evo" => "Evolutionary algorithms — population, evolve, fitness",
        "rl" => "Reinforcement learning — Q-learning, DQN, policy",
        "monitor" => "System monitoring — metrics, alerts, dashboards",
        "cloud" => "Cloud platform — deploy, instances, regions",
        "repl" => "Remote REPL — serve, connect, broadcast via WebSocket",
        "workspace" => "Team workspaces — create, share agents, collaborate",
        "marketplace" => "Plugin marketplace — publish, search, install",
        "telemetry" => "Usage telemetry — opt-in analytics and reporting",
        "docker" => "Docker — containers, images, compose",
        "podman" => "Podman — rootless containers",
        "container" => "Container abstraction — runtime-agnostic container ops",
        "k8s" => "Kubernetes — pods, services, deployments",
        "helm" => "Helm — chart management and releases",
        "vm" => "Virtual machines — create, start, stop, snapshot",
        "hyperv" => "Hyper-V — Windows VM management",
        "virsh" => "Libvirt — Linux VM management",
        "wsl" => "WSL — Windows Subsystem for Linux",
        _ => "Module functions",
    }
}

/// Get module definitions from the module system for the ontology
fn get_module_definitions() -> Vec<ModuleDefinition> {
    let all = all_modules();
    let mut modules = Vec::new();

    for (name, value) in &all {
        if let Value::Record(map) = value {
            let mut functions = Vec::new();
            for (method, _builtin_val) in map {
                // Extract the builtin name from the Value::Builtin
                let builtin_name = if let Value::Builtin(bref) = _builtin_val {
                    bref.name.clone()
                } else {
                    method.clone()
                };
                functions.push(ModuleFunctionDef {
                    method: method.clone(),
                    builtin: builtin_name.clone(),
                    syntax: format!("{}.{}()", name, method),
                });
            }
            functions.sort_by(|a, b| a.method.cmp(&b.method));

            modules.push(ModuleDefinition {
                name: name.to_string(),
                description: module_description(name).to_string(),
                functions,
            });
        }
    }

    // Deduplicate modules (e.g., "platform" appears 3 times in all_modules)
    modules.sort_by(|a, b| a.name.cmp(&b.name));
    modules.dedup_by(|a, b| a.name == b.name);

    modules
}

/// Get operator definitions
fn get_operator_definitions() -> Vec<OperatorDefinition> {
    vec![
        OperatorDefinition {
            symbol: "|".to_string(),
            name: "pipe".to_string(),
            description: "Pipeline operator - pass left result to right function".to_string(),
            precedence: 1,
            associativity: "left".to_string(),
            operand_types: vec!["Value".to_string(), "Function".to_string()],
        },
        OperatorDefinition {
            symbol: "+".to_string(),
            name: "add".to_string(),
            description: "Addition or string concatenation".to_string(),
            precedence: 5,
            associativity: "left".to_string(),
            operand_types: vec!["Number".to_string(), "String".to_string()],
        },
        OperatorDefinition {
            symbol: "-".to_string(),
            name: "subtract".to_string(),
            description: "Subtraction".to_string(),
            precedence: 5,
            associativity: "left".to_string(),
            operand_types: vec!["Number".to_string()],
        },
        OperatorDefinition {
            symbol: "*".to_string(),
            name: "multiply".to_string(),
            description: "Multiplication".to_string(),
            precedence: 6,
            associativity: "left".to_string(),
            operand_types: vec!["Number".to_string()],
        },
        OperatorDefinition {
            symbol: "/".to_string(),
            name: "divide".to_string(),
            description: "Division".to_string(),
            precedence: 6,
            associativity: "left".to_string(),
            operand_types: vec!["Number".to_string()],
        },
        OperatorDefinition {
            symbol: "==".to_string(),
            name: "equals".to_string(),
            description: "Equality comparison".to_string(),
            precedence: 3,
            associativity: "left".to_string(),
            operand_types: vec!["Value".to_string()],
        },
        OperatorDefinition {
            symbol: "!=".to_string(),
            name: "not_equals".to_string(),
            description: "Inequality comparison".to_string(),
            precedence: 3,
            associativity: "left".to_string(),
            operand_types: vec!["Value".to_string()],
        },
        OperatorDefinition {
            symbol: ">".to_string(),
            name: "greater_than".to_string(),
            description: "Greater than comparison".to_string(),
            precedence: 4,
            associativity: "left".to_string(),
            operand_types: vec!["Number".to_string()],
        },
        OperatorDefinition {
            symbol: "<".to_string(),
            name: "less_than".to_string(),
            description: "Less than comparison".to_string(),
            precedence: 4,
            associativity: "left".to_string(),
            operand_types: vec!["Number".to_string()],
        },
        OperatorDefinition {
            symbol: "&&".to_string(),
            name: "and".to_string(),
            description: "Logical AND".to_string(),
            precedence: 2,
            associativity: "left".to_string(),
            operand_types: vec!["Bool".to_string()],
        },
        OperatorDefinition {
            symbol: "||".to_string(),
            name: "or".to_string(),
            description: "Logical OR".to_string(),
            precedence: 2,
            associativity: "left".to_string(),
            operand_types: vec!["Bool".to_string()],
        },
        OperatorDefinition {
            symbol: ".".to_string(),
            name: "field_access".to_string(),
            description: "Access record field".to_string(),
            precedence: 10,
            associativity: "left".to_string(),
            operand_types: vec!["Record".to_string(), "String".to_string()],
        },
    ]
}

/// Get category information
fn get_category_info() -> Vec<CategoryInfo> {
    let builtins = get_all_builtin_definitions();
    let mut categories: HashMap<String, usize> = HashMap::new();

    for b in &builtins {
        *categories.entry(b.category.clone()).or_insert(0) += 1;
    }

    let descriptions: HashMap<&str, &str> = [
        ("FileSystem", "File and directory operations"),
        (
            "Functional",
            "Higher-order functions for data transformation",
        ),
        ("Aggregation", "Statistical and aggregation functions"),
        ("String", "String manipulation functions"),
        ("Array", "Array and collection operations"),
        ("Network", "HTTP and network operations"),
        ("AI", "AI model queries and agent operations"),
        ("MCP", "Model Context Protocol tools"),
        ("NeuralNet", "Neural network creation and training"),
        ("Evolution", "Evolutionary algorithm operations"),
        ("ReinforcementLearning", "Reinforcement learning agents"),
        ("Knowledge", "Knowledge graphs and RAG operations"),
        ("System", "System info — hostname, OS, CPU, memory"),
        ("Process", "Process management — list, kill, monitor"),
        ("Service", "Service/daemon management"),
        ("Database", "Database queries and connections"),
        ("Crypto", "Cryptography, hashing, and encryption"),
        ("Container", "Docker/Podman container management"),
        ("Kubernetes", "Kubernetes cluster operations"),
        ("Virtualization", "VM and hypervisor management"),
        ("Cloud", "Cloud infrastructure and IaC"),
        ("RemoteAccess", "SSH, SCP, and remote connectivity"),
        ("Security", "Firewall, SELinux, AppArmor"),
        ("Monitoring", "System monitoring and performance"),
        ("GUI", "GUI automation — screenshots, clicks"),
        ("Web", "Web automation and browser control"),
        ("Archive", "Compression and archive operations"),
        ("Scheduling", "Cron jobs and scheduled tasks"),
        ("UserManagement", "User and group management"),
        ("Package", "Package management"),
        ("Hardware", "Hardware info and device access"),
        ("Input", "Clipboard and user input"),
        ("JSON", "JSON parsing and manipulation"),
        ("Math", "Mathematical operations and constants"),
        ("Shell", "Shell operations and command execution"),
        ("Git", "Git version control operations"),
        ("Code", "Code intelligence and analysis"),
        ("Project", "Project scaffolding and management"),
        ("Search", "Code search and symbol lookup"),
        ("Testing", "Test execution and analysis"),
        ("Diagnostics", "Linting, profiling, type checking"),
        ("Refactoring", "Code refactoring operations"),
        ("Session", "Session save/restore and history"),
        ("Documentation", "Documentation generation and browsing"),
        ("DevEnvironment", "Development environment setup"),
        ("Platform", "Platform detection and capabilities"),
        ("A2A", "Agent-to-Agent messaging protocol"),
        ("A2UI", "Agent-to-UI notification protocol"),
        ("NANDA", "NANDA consensus protocol"),
        ("RBAC", "Role-based access control"),
        ("Audit", "Audit logging and compliance"),
        ("SSO", "Single sign-on authentication"),
        ("Cluster", "Distributed computing and job scheduling"),
        ("REPL", "Remote REPL and WebSocket sessions"),
        ("Workspace", "Team workspace management"),
        ("Marketplace", "Plugin marketplace"),
        ("Telemetry", "Usage telemetry and analytics"),
        ("Core", "Core language utilities and builtins"),
    ]
    .into_iter()
    .collect();

    let mut result: Vec<CategoryInfo> = categories
        .into_iter()
        .map(|(name, count)| CategoryInfo {
            description: descriptions.get(name.as_str()).unwrap_or(&"").to_string(),
            name,
            builtin_count: count,
        })
        .collect();
    result.sort_by(|a, b| a.name.cmp(&b.name));
    result
}

// ============================================================================
// AI Provider Schema Formats
// ============================================================================

use crate::providers::schema::{builtins_to_tools, ToolFormat};

// ----------------------------------------------------------------------------
// IronShell types as JSON Schema
// ----------------------------------------------------------------------------
//
// The catalogue's parameter types are IronShell's own words (`String`, `Int`,
// `Record`, `Array | String`, `Lambda`). Provider function schemas accept JSON
// Schema only, so every `"type"` the provider formats carry is mapped here. The
// `ontology` format keeps IronShell's names: they are its vocabulary.

/// One piece of an IronShell type, in JSON Schema terms.
enum SchemaPiece {
    /// A JSON Schema type, with the schema of its elements for an array.
    Type(&'static str, Option<JsonValue>),
    /// Any value: JSON Schema says so by having no `type` at all.
    Any,
    /// A lambda cannot be a JSON value, so it is passed as source text.
    Lambda,
}

fn schema_piece(part: &str) -> SchemaPiece {
    let part = part.trim();
    let lower = part.to_ascii_lowercase();
    // `Array<Number>` and `[Number]` say what the elements are.
    let element = lower
        .strip_prefix("array<")
        .and_then(|r| r.strip_suffix('>'))
        .or_else(|| lower.strip_prefix('[').and_then(|r| r.strip_suffix(']')));
    if let Some(element) = element {
        let items = match schema_piece(element) {
            SchemaPiece::Type(t, items) => {
                let mut s = json!({ "type": t });
                if let Some(items) = items {
                    s["items"] = items;
                }
                Some(s)
            }
            SchemaPiece::Lambda => Some(json!({ "type": "string" })),
            SchemaPiece::Any => None,
        };
        return SchemaPiece::Type("array", items);
    }
    match lower.as_str() {
        "string" | "str" | "path" | "text" | "uri" | "url" => SchemaPiece::Type("string", None),
        "int" | "integer" => SchemaPiece::Type("integer", None),
        "float" | "number" | "numeric" => SchemaPiece::Type("number", None),
        "bool" | "boolean" => SchemaPiece::Type("boolean", None),
        "array" | "list" => SchemaPiece::Type("array", None),
        "record" | "object" | "table" | "map" => SchemaPiece::Type("object", None),
        "null" | "none" => SchemaPiece::Type("null", None),
        "lambda" | "fn" | "function" => SchemaPiece::Lambda,
        // `Any`, `Value`, and a word that is no JSON type: any value is accepted.
        _ => SchemaPiece::Any,
    }
}

/// Rewrite every `"type"` in a schema from an IronShell type name to JSON
/// Schema. `Any` loses its `type`; a `Lambda` becomes a string whose
/// description says it is IronShell lambda source; a union (`Array | String`)
/// becomes a list of types. Keys that start with `x-` are IronShell's own and
/// are left alone. Applying it twice changes nothing.
fn normalize_schema_types(schema: &mut JsonValue) {
    match schema {
        JsonValue::Object(map) => {
            if let Some(JsonValue::String(ty)) = map.get("type").cloned() {
                let mut types: Vec<&'static str> = Vec::new();
                let mut items: Option<JsonValue> = None;
                let (mut any, mut lambda) = (false, false);
                for piece in ty.split('|').map(schema_piece) {
                    match piece {
                        SchemaPiece::Type(t, i) => {
                            if !types.contains(&t) {
                                types.push(t);
                            }
                            items = i;
                        }
                        SchemaPiece::Lambda => {
                            lambda = true;
                            if !types.contains(&"string") {
                                types.push("string");
                            }
                        }
                        SchemaPiece::Any => any = true,
                    }
                }
                if any || types.is_empty() {
                    map.remove("type");
                } else if types.len() == 1 {
                    map.insert("type".to_string(), json!(types[0]));
                    if let (Some(items), false) = (items, map.contains_key("items")) {
                        map.insert("items".to_string(), items);
                    }
                } else {
                    map.insert("type".to_string(), json!(types));
                }
                if lambda {
                    let note = "IronShell lambda source, e.g. fn(x) => x * 2";
                    let described = match map.get("description").and_then(JsonValue::as_str) {
                        Some(d) if !d.is_empty() => format!("{d} ({note})"),
                        _ => note.to_string(),
                    };
                    map.insert("description".to_string(), json!(described));
                }
            }
            for (key, child) in map.iter_mut() {
                if !key.starts_with("x-") {
                    normalize_schema_types(child);
                }
            }
        }
        JsonValue::Array(items) => items.iter_mut().for_each(normalize_schema_types),
        _ => {}
    }
}

/// A schema in the subset Gemini's function declarations document: upper-case
/// type names, no list-valued `type` (a union is `anyOf`), no `x-` keys.
fn gemini_schema(schema: &JsonValue) -> JsonValue {
    match schema {
        JsonValue::Object(map) => {
            let mut out = serde_json::Map::new();
            for (key, value) in map {
                if key.starts_with("x-") {
                    continue;
                }
                match (key.as_str(), value) {
                    ("type", JsonValue::String(t)) => {
                        out.insert(key.clone(), json!(t.to_ascii_uppercase()));
                    }
                    ("type", JsonValue::Array(types)) => {
                        let variants: Vec<JsonValue> = types
                            .iter()
                            .filter_map(JsonValue::as_str)
                            .map(|t| json!({ "type": t.to_ascii_uppercase() }))
                            .collect();
                        out.insert("anyOf".to_string(), JsonValue::Array(variants));
                    }
                    ("properties", JsonValue::Object(props)) => {
                        // A property is a schema, whatever it is called.
                        let props = props
                            .iter()
                            .map(|(k, v)| (k.clone(), gemini_schema(v)))
                            .collect();
                        out.insert(key.clone(), JsonValue::Object(props));
                    }
                    _ => {
                        out.insert(key.clone(), gemini_schema(value));
                    }
                }
            }
            JsonValue::Object(out)
        }
        JsonValue::Array(items) => JsonValue::Array(items.iter().map(gemini_schema).collect()),
        other => other.clone(),
    }
}

/// OpenAI refuses a function description over 1,024 characters. A longer one
/// is cut at a word boundary and ends in an ellipsis, and the result fits the
/// limit.
fn truncate_description(text: &str, max: usize) -> String {
    if text.chars().count() <= max {
        return text.to_string();
    }
    let kept: String = text.chars().take(max.saturating_sub(1)).collect();
    let cut = match kept.rfind(char::is_whitespace) {
        Some(i) if i > 0 => &kept[..i],
        _ => kept.as_str(),
    };
    format!("{}\u{2026}", cut.trim_end())
}

/// Convert ontology builtins to tool list in the specified format with optional name prefix
fn ontology_tools(ontology: &LanguageOntology, format: ToolFormat, prefix: &str) -> Vec<JsonValue> {
    // OpenAI-shaped tools cap a function description at 1,024 characters.
    let descriptions: Vec<String> = ontology
        .builtins
        .iter()
        .map(|b| {
            if format == ToolFormat::OpenAI {
                truncate_description(&b.description, 1024)
            } else {
                b.description.clone()
            }
        })
        .collect();
    builtins_to_tools(
        ontology
            .builtins
            .iter()
            .zip(descriptions.iter())
            .map(|(b, d)| (b.name.as_str(), d.as_str(), &b.json_schema)),
        format,
        prefix,
    )
}

/// Build OpenAI function calling schema
fn build_openai_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "ironshell_");

    json!({
        "format": "openai_function_calling",
        "version": "2025-01",
        "compatible_models": [
            "gpt-5", "gpt-5-mini", "gpt-5-turbo",
            "gpt-4o", "gpt-4o-mini", "gpt-4-turbo",
            "o1", "o1-mini", "o1-preview",
            "o3", "o3-mini", "o4-mini"
        ],
        "tools": tools,
        "instructions": "Use these tools to execute IronShell operations. Each tool corresponds to an IronShell builtin function."
    })
}

/// Build Anthropic Claude tool use schema
fn build_claude_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::Anthropic, "ironshell_");

    json!({
        "format": "anthropic_tool_use",
        "version": "2025-01",
        "compatible_models": [
            "claude-4.5-opus", "claude-4.5-sonnet", "claude-4.5-haiku",
            "claude-4-opus", "claude-4-sonnet", "claude-4-haiku",
            "claude-3.5-sonnet-v2", "claude-3.5-haiku",
            "claude-3-opus", "claude-3-sonnet", "claude-3-haiku"
        ],
        "tools": tools,
        "instructions": "Use these tools to execute IronShell operations. Results are returned as JSON."
    })
}

/// Build Google Gemini function declaration schema
fn build_gemini_schema(ontology: &LanguageOntology) -> JsonValue {
    // Gemini documents its own subset of the schema: upper-case types, `anyOf`
    // for a union, no extension keys, and no `parameters` for a function that
    // takes none (an OBJECT with no properties is refused).
    let function_declarations: Vec<JsonValue> =
        ontology_tools(ontology, ToolFormat::Google, "ironshell_")
            .into_iter()
            .map(|mut declaration| {
                if let Some(obj) = declaration.as_object_mut() {
                    let no_params = obj
                        .get("parameters")
                        .and_then(|p| p.get("properties"))
                        .and_then(JsonValue::as_object)
                        .is_none_or(|props| props.is_empty());
                    match obj.remove("parameters") {
                        Some(params) if !no_params => {
                            obj.insert("parameters".to_string(), gemini_schema(&params));
                        }
                        _ => {}
                    }
                }
                declaration
            })
            .collect();

    json!({
        "format": "gemini_function_calling",
        "version": "v2",
        "compatible_models": [
            "gemini-2.5-pro", "gemini-2.5-flash",
            "gemini-2.0-flash", "gemini-2.0-flash-thinking",
            "gemini-1.5-pro", "gemini-1.5-flash"
        ],
        "function_declarations": function_declarations,
        "instructions": "Use these functions to execute IronShell operations."
    })
}

/// Build compact ontology for context efficiency
fn build_compact_ontology(ontology: &LanguageOntology) -> JsonValue {
    // Compact format optimized for LLM context windows
    let builtins_compact: Vec<JsonValue> = ontology
        .builtins
        .iter()
        .map(|b| {
            let mut entry = json!({
                "n": b.name,
                "d": b.description,
                "s": b.signature,
                "c": b.category
            });
            if let Some(aliases) = &b.aliases {
                if !aliases.is_empty() {
                    entry["a"] = json!(aliases);
                }
            }
            entry
        })
        .collect();

    let modules_compact: Vec<JsonValue> = ontology
        .modules
        .iter()
        .map(|m| {
            let fns: Vec<String> = m.functions.iter().map(|f| f.method.clone()).collect();
            json!({
                "n": m.name,
                "d": m.description,
                "fns": fns
            })
        })
        .collect();

    json!({
        "lang": "IronShell",
        "ver": ontology.language.version,
        "ontology_ver": ontology.ontology_version,
        "types": ["Int", "Float", "String", "Bool", "Array", "Record", "Lambda", "Table", "Option"],
        "ops": ["|", "+", "-", "*", "/", "%", "^", "==", "!=", ">", "<", ">=", "<=", "&&", "||", "!", "."],
        "syntax": {
            "let": "let x = expr",
            "fn": "fn(x) => expr",
            "pipe": "a | b | c",
            "rec": "{k: v}",
            "arr": "[a, b]",
            "if": "if cond { a } else { b }",
            "match": "match x { p => r }",
            "import": IMPORT_SYNTAX
        },
        "modules": modules_compact,
        "builtins": builtins_compact,
        "stats": {
            "total_builtins": ontology.builtins.len(),
            "total_modules": ontology.modules.len(),
            "categories": ontology.categories.len()
        },
        "os_ontology": ontology.os_ontology,
        "cli_tools": ontology.cli_tools,
    })
}

/// Build Meta Llama function calling schema
fn build_llama_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "ironshell_");

    json!({
        "format": "llama_function_calling",
        "version": "4.0",
        "compatible_models": [
            "llama-4-maverick-17b", "llama-4-maverick-70b", "llama-4-maverick-405b",
            "llama-4-scout-17b", "llama-4-scout-109b",
            "llama-3.3-70b", "llama-3.2-90b-vision", "llama-3.2-11b-vision",
            "llama-3.1-405b", "llama-3.1-70b", "llama-3.1-8b"
        ],
        "tools": tools,
        "system_prompt": "You are an assistant with access to IronShell tools. Use them to help users with shell operations, data processing, and AI tasks.",
        "instructions": "Call tools using JSON format. Tool names are prefixed with 'ironshell_'."
    })
}

/// Build Mistral AI function calling schema
fn build_mistral_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "mistral_function_calling",
        "version": "v2",
        "compatible_models": [
            "mistral-large-2501", "mistral-large-latest",
            "mistral-medium-2501", "mistral-small-2501",
            "codestral-2501", "codestral-latest",
            "pixtral-large-2501", "pixtral-12b",
            "ministral-8b", "ministral-3b"
        ],
        "tools": tools,
        "tool_choice": "auto",
        "instructions": "Use these tools to execute IronShell operations. Results are returned as JSON."
    })
}

/// Build Cohere Command R function calling schema
fn build_cohere_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::Cohere, "");

    json!({
        "format": "cohere_tools",
        "version": "v3",
        "compatible_models": [
            "command-a", "command-a-03-2025",
            "command-r-plus", "command-r-plus-08-2024",
            "command-r", "command-r7b",
            "command-nightly"
        ],
        "tools": tools,
        "instructions": "Use these tools to execute IronShell operations. Call tools by name with appropriate parameters."
    })
}

/// Build xAI Grok function calling schema
fn build_grok_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "grok_function_calling",
        "version": "v2",
        "compatible_models": [
            "grok-3", "grok-3-mini", "grok-3-fast",
            "grok-2", "grok-2-mini", "grok-2-vision",
            "grok-2-1212"
        ],
        "tools": tools,
        "instructions": "Use these tools to execute IronShell operations. Results are returned as JSON."
    })
}

/// Build DeepSeek function calling schema
fn build_deepseek_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "deepseek_function_calling",
        "version": "v4",
        "compatible_models": [
            "deepseek-chat", "deepseek-reasoner",
            "deepseek-v3", "deepseek-v3-0324",
            "deepseek-r1", "deepseek-r1-0528",
            "deepseek-r1-distill-llama-70b", "deepseek-r1-distill-qwen-32b",
            "deepseek-coder-v2", "deepseek-coder-v2.5"
        ],
        "tools": tools,
        "reasoning_support": true,
        "instructions": "Use these tools to execute IronShell operations. For complex tasks, use chain-of-thought reasoning."
    })
}

/// Build AWS Bedrock converse API schema
fn build_bedrock_schema(ontology: &LanguageOntology) -> JsonValue {
    // Bedrock uses Anthropic-style input_schema wrapped in toolSpec/inputSchema/json
    let tool_config: Vec<JsonValue> = ontology
        .builtins
        .iter()
        .map(|b| {
            json!({
                "toolSpec": {
                    "name": b.name,
                    "description": b.description,
                    "inputSchema": {
                        "json": b.json_schema
                    }
                }
            })
        })
        .collect();

    json!({
        "format": "bedrock_converse",
        "version": "v2",
        "compatible_models": [
            "anthropic.claude-4-5-opus-20250115-v1:0",
            "anthropic.claude-4-5-sonnet-20250115-v1:0",
            "anthropic.claude-4-5-haiku-20250115-v1:0",
            "anthropic.claude-3-5-sonnet-20241022-v2:0",
            "anthropic.claude-3-5-haiku-20241022-v1:0",
            "anthropic.claude-3-opus-20240229-v1:0",
            "amazon.nova-pro-v1:0",
            "amazon.nova-lite-v1:0",
            "amazon.nova-micro-v1:0",
            "meta.llama3-3-70b-instruct-v1:0",
            "meta.llama3-1-405b-instruct-v1:0",
            "mistral.mistral-large-2411-v1:0"
        ],
        "toolConfig": {
            "tools": tool_config
        },
        "instructions": "Use toolUse blocks in your response to invoke IronShell tools."
    })
}

/// Build Azure OpenAI function calling schema
fn build_azure_openai_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "azure_openai_function_calling",
        "version": "2025-01-01-preview",
        "compatible_deployments": [
            "gpt-5", "gpt-5-mini", "gpt-5-turbo",
            "gpt-4o", "gpt-4o-mini", "gpt-4-turbo", "gpt-4", "gpt-35-turbo",
            "o1", "o1-preview", "o1-mini", "o3-mini", "o4-mini"
        ],
        "tools": tools,
        "api_version": "2025-01-01-preview",
        "instructions": "Use these tools via Azure OpenAI API. Same format as OpenAI but with Azure endpoint."
    })
}

/// Build Alibaba Qwen function calling schema
fn build_qwen_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "qwen_function_calling",
        "version": "v3",
        "compatible_models": [
            "qwen3-235b-a22b", "qwen3-32b", "qwen3-14b", "qwen3-8b", "qwen3-4b",
            "qwen2.5-max", "qwen2.5-plus", "qwen2.5-turbo",
            "qwen2.5-72b-instruct", "qwen2.5-coder-32b-instruct",
            "qwq-32b", "qvq-72b-preview"
        ],
        "tools": tools,
        "instructions": "Use these tools to execute IronShell operations. Compatible with DashScope API."
    })
}

/// Build Ollama local model schema
fn build_ollama_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "ollama_tools",
        "version": "v0.5",
        "compatible_models": [
            "llama4:scout-17b", "llama4:maverick-17b",
            "llama3.3:70b", "llama3.2:3b", "llama3.1:8b", "llama3.1:70b",
            "qwen3:32b", "qwen3:8b", "qwen2.5:72b", "qwen2.5-coder:32b",
            "mistral-large:123b", "codestral:22b",
            "deepseek-r1:70b", "deepseek-r1:32b", "deepseek-r1:14b", "deepseek-r1:7b",
            "gemma3:27b", "phi4:14b", "command-a:111b"
        ],
        "tools": tools,
        "endpoint": "http://localhost:11434/api/chat",
        "instructions": "Use these tools with Ollama's chat API. Requires tool-capable model."
    })
}

/// Build vLLM OpenAI-compatible schema
fn build_vllm_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "vllm_openai_compatible",
        "version": "v1",
        "compatible_models": ["Any tool-capable model served via vLLM"],
        "tools": tools,
        "endpoint": "/v1/chat/completions",
        "tool_choice_modes": ["auto", "required", "none", {"type": "function", "function": {"name": "specific_tool"}}],
        "instructions": "Use OpenAI-compatible format with vLLM server. Ensure model supports function calling."
    })
}

/// Build HuggingFace Inference Endpoints schema
fn build_huggingface_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "huggingface_tgi",
        "version": "v3",
        "compatible_endpoints": ["inference-endpoints", "text-generation-inference", "transformers"],
        "recommended_models": [
            "meta-llama/Llama-4-Maverick-17B-128E-Instruct",
            "meta-llama/Llama-3.3-70B-Instruct",
            "Qwen/Qwen3-235B-A22B",
            "Qwen/Qwen2.5-72B-Instruct",
            "mistralai/Mistral-Large-Instruct-2501",
            "deepseek-ai/DeepSeek-V3",
            "deepseek-ai/DeepSeek-R1",
            "google/gemma-3-27b-it"
        ],
        "tools": tools,
        "instructions": "Use with HuggingFace Inference Endpoints or local TGI deployment."
    })
}

/// Build OpenRouter unified schema
fn build_openrouter_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "openrouter_unified",
        "version": "v2",
        "tool_capable_models": [
            "openai/gpt-5", "openai/gpt-5-mini", "openai/gpt-4o", "openai/gpt-4o-mini", "openai/o3", "openai/o4-mini",
            "anthropic/claude-4.5-opus", "anthropic/claude-4.5-sonnet", "anthropic/claude-4.5-haiku",
            "anthropic/claude-3.5-sonnet", "anthropic/claude-3-opus",
            "google/gemini-2.5-pro", "google/gemini-2.0-flash",
            "meta-llama/llama-4-maverick-70b", "meta-llama/llama-3.3-70b-instruct",
            "mistralai/mistral-large-2501", "mistralai/codestral-latest",
            "deepseek/deepseek-v3", "deepseek/deepseek-r1",
            "qwen/qwen-2.5-72b-instruct", "qwen/qwq-32b-preview",
            "x-ai/grok-3", "cohere/command-a"
        ],
        "tools": tools,
        "api_base": "https://openrouter.ai/api/v1",
        "instructions": "Use OpenAI-compatible format. OpenRouter will route to the specified model provider."
    })
}

/// Build Moonshot AI Kimi function calling schema
fn build_kimi_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "kimi_function_calling",
        "version": "v2",
        "compatible_models": [
            "kimi-latest", "kimi-k2",
            "moonshot-v1-128k", "moonshot-v1-32k", "moonshot-v1-8k",
            "moonshot-v2-128k"
        ],
        "tools": tools,
        "api_base": "https://api.moonshot.cn/v1",
        "instructions": "Use OpenAI-compatible format with Moonshot API. Kimi excels at long-context tasks."
    })
}

/// Build 01.AI Yi function calling schema
fn build_yi_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "yi_function_calling",
        "version": "v2",
        "compatible_models": [
            "yi-large-fc", "yi-large-turbo", "yi-large-preview",
            "yi-lightning", "yi-lightning-lite",
            "yi-medium-200k", "yi-vision-v2",
            "yi-coder-9b", "yi-coder-1.5b"
        ],
        "tools": tools,
        "api_base": "https://api.01.ai/v1",
        "instructions": "Use OpenAI-compatible format with 01.AI API. Yi models support function calling."
    })
}

/// Build Zhipu AI GLM function calling schema
fn build_glm_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "glm_function_calling",
        "version": "v5",
        "compatible_models": [
            "glm-5", "glm-5-plus", "glm-5-air",
            "glm-4-alltools", "glm-4-plus", "glm-4-air", "glm-4-airx", "glm-4-flash",
            "glm-4v-plus", "glm-4v-flash",
            "codegeex-4", "cogvideo-x"
        ],
        "tools": tools,
        "api_base": "https://open.bigmodel.cn/api/paas/v4",
        "instructions": "Use OpenAI-compatible format with Zhipu API. GLM-5 supports advanced tools and code generation."
    })
}

/// Build Reka AI function calling schema
fn build_reka_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "reka_function_calling",
        "version": "v2",
        "compatible_models": [
            "reka-core-20250115", "reka-core",
            "reka-flash-20250115", "reka-flash",
            "reka-edge-20250115", "reka-edge",
            "reka-vibe"
        ],
        "tools": tools,
        "api_base": "https://api.reka.ai/v1",
        "multimodal": true,
        "instructions": "Use OpenAI-compatible format with Reka API. Reka models are natively multimodal."
    })
}

/// Build AI21 Labs function calling schema
fn build_ai21_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "ai21_function_calling",
        "version": "v2",
        "compatible_models": [
            "jamba-2-large", "jamba-2-mini",
            "jamba-1.5-large", "jamba-1.5-mini",
            "jamba-instruct"
        ],
        "tools": tools,
        "api_base": "https://api.ai21.com/studio/v1",
        "instructions": "Use OpenAI-compatible format with AI21 API. Jamba 2 models have 256K context and hybrid architecture."
    })
}

/// Build Perplexity AI function calling schema
fn build_perplexity_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "perplexity_function_calling",
        "version": "v2",
        "compatible_models": [
            "sonar-pro", "sonar",
            "sonar-reasoning-pro", "sonar-reasoning",
            "sonar-deep-research",
            "r1-1776"
        ],
        "tools": tools,
        "api_base": "https://api.perplexity.ai",
        "online_search": true,
        "instructions": "Use OpenAI-compatible format with Perplexity API. Sonar models have built-in web search."
    })
}

/// Build Together AI function calling schema
fn build_together_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "together_function_calling",
        "version": "v2",
        "tool_capable_models": [
            "meta-llama/Llama-4-Maverick-17B-128E-Instruct-FP8",
            "meta-llama/Llama-3.3-70B-Instruct-Turbo",
            "meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo",
            "Qwen/Qwen3-235B-A22B-fp8-tput",
            "Qwen/Qwen2.5-72B-Instruct-Turbo",
            "deepseek-ai/DeepSeek-R1",
            "deepseek-ai/DeepSeek-V3",
            "mistralai/Mistral-Large-2501-Instruct-FP8",
            "google/gemma-3-27b-it"
        ],
        "tools": tools,
        "api_base": "https://api.together.xyz/v1",
        "instructions": "Use OpenAI-compatible format with Together API. Supports many open-source models."
    })
}

/// Build Groq function calling schema (distinct from xAI Grok)
fn build_groq_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "groq_function_calling",
        "version": "v2",
        "compatible_models": [
            "llama-4-scout-17b-16e-instruct",
            "llama-3.3-70b-versatile", "llama-3.3-70b-specdec",
            "llama-3.1-70b-versatile", "llama-3.1-8b-instant",
            "qwen-qwq-32b", "deepseek-r1-distill-llama-70b",
            "mixtral-8x7b-32768", "gemma2-9b-it"
        ],
        "tools": tools,
        "api_base": "https://api.groq.com/openai/v1",
        "ultra_low_latency": true,
        "instructions": "Use OpenAI-compatible format with Groq API. Ultra-fast inference on LPU hardware."
    })
}

/// Build Fireworks AI function calling schema
fn build_fireworks_schema(ontology: &LanguageOntology) -> JsonValue {
    let tools = ontology_tools(ontology, ToolFormat::OpenAI, "");

    json!({
        "format": "fireworks_function_calling",
        "version": "v2",
        "tool_capable_models": [
            "accounts/fireworks/models/llama4-maverick-instruct-basic",
            "accounts/fireworks/models/llama-v3p3-70b-instruct",
            "accounts/fireworks/models/llama-v3p1-405b-instruct",
            "accounts/fireworks/models/qwen3-235b-a22b",
            "accounts/fireworks/models/qwen2p5-72b-instruct",
            "accounts/fireworks/models/deepseek-r1",
            "accounts/fireworks/models/deepseek-v3",
            "accounts/fireworks/models/mistral-large-2501-instruct"
        ],
        "tools": tools,
        "api_base": "https://api.fireworks.ai/inference/v1",
        "fast_inference": true,
        "instructions": "Use OpenAI-compatible format with Fireworks API. Optimized for fast model inference."
    })
}

// ============================================================================
// HTTP Server Integration
// ============================================================================

#[cfg(feature = "native")]
pub mod server {
    use super::*;
    use axum::{
        body::Body,
        extract::ws::{Message, WebSocket, WebSocketUpgrade},
        http::{header, StatusCode},
        response::{IntoResponse, Response},
        routing::{get, post},
        Json, Router,
    };
    use std::collections::HashMap;
    use std::convert::Infallible;
    use std::sync::Arc;
    use std::time::Duration;
    use tokio::sync::{broadcast, mpsc, RwLock};
    use tokio::task::JoinHandle;
    use tower_http::cors::{Any, CorsLayer};
    use tower_http::timeout::TimeoutLayer;

    /// Requests handled at once across the whole server. Each in-flight
    /// handler holds one permit until it returns its response, so a flood of
    /// slow evaluations queues here instead of filling the blocking pool.
    /// Streaming and WebSocket responses release their permit as soon as the
    /// response head is produced, so long-lived connections do not hold slots.
    const MAX_CONCURRENT_REQUESTS: usize = 64;

    /// How long a request waits for one of those slots before it is answered
    /// 503. The queue had no bound in time: with every slot held by a slow
    /// evaluation, a client waited as long as the slowest of them, and learned
    /// nothing about why.
    const MAX_QUEUE_WAIT: Duration = Duration::from_secs(30);

    /// Evaluations in flight on the streaming (SSE) and WebSocket routes at
    /// once. Those routes release their `admit` slot as soon as the response
    /// head exists, so without this bound a client could open any number of
    /// streams and run an evaluation in each.
    const MAX_CONCURRENT_STREAMS: usize = 16;

    /// The stream slots, shared by every SSE and WebSocket evaluation in this
    /// process.
    fn stream_slots() -> Arc<tokio::sync::Semaphore> {
        static SLOTS: std::sync::OnceLock<Arc<tokio::sync::Semaphore>> = std::sync::OnceLock::new();
        SLOTS
            .get_or_init(|| Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_STREAMS)))
            .clone()
    }

    /// Take a stream slot without waiting, or `None` when all are held. A
    /// refusal is immediate: a streamed evaluation that queued would hold its
    /// connection open for the wait as well as the work.
    fn try_take_stream_slot(
        slots: &Arc<tokio::sync::Semaphore>,
    ) -> Option<tokio::sync::OwnedSemaphorePermit> {
        slots.clone().try_acquire_owned().ok()
    }

    fn stream_busy_detail() -> String {
        format!("all {MAX_CONCURRENT_STREAMS} streaming evaluation slots are in use; retry later")
    }

    /// The 503 an SSE route answers when no stream slot is free.
    fn stream_busy_response() -> Response {
        (
            StatusCode::SERVICE_UNAVAILABLE,
            [(header::RETRY_AFTER, "5")],
            Json(serde_json::json!({
                "error": "busy",
                "code": "E_BUDGET_EXCEEDED",
                "detail": stream_busy_detail(),
            })),
        )
            .into_response()
    }

    /// The response a WebSocket `execute` gets when no stream slot is free.
    fn ws_busy_response() -> AgentResponse {
        AgentResponse {
            code: Some("E_BUDGET_EXCEEDED".to_string()),
            hint: Some(
                "retry after an in-flight streamed or WebSocket evaluation finishes".to_string(),
            ),
            success: false,
            result: None,
            error: Some(stream_busy_detail()),
            result_type: None,
            metadata: None,
        }
    }

    /// Admits a request once one of the `max` slots in `permits` is free,
    /// holding it until the response head is produced, or answers 503 after
    /// `wait`. One semaphore is shared by every route.
    pub(crate) async fn admit(
        permits: Arc<tokio::sync::Semaphore>,
        max: usize,
        wait: Duration,
        req: axum::extract::Request,
        next: axum::middleware::Next,
    ) -> axum::response::Response {
        match tokio::time::timeout(wait, permits.acquire_owned()).await {
            Ok(Ok(permit)) => {
                let res = next.run(req).await;
                drop(permit);
                res
            }
            _ => (
                StatusCode::SERVICE_UNAVAILABLE,
                [(header::RETRY_AFTER, "5")],
                Json(serde_json::json!({
                    "error": "busy",
                    "code": "E_BUDGET_EXCEEDED",
                    "detail": format!(
                        "all {max} request slots stayed busy for {} ms; retry later",
                        wait.as_millis()
                    ),
                })),
            )
                .into_response(),
        }
    }

    // ========================================================================
    // WebSocket Types
    // ========================================================================

    /// WebSocket message from client
    #[derive(Debug, Clone, serde::Deserialize)]
    #[serde(tag = "type")]
    pub enum WsClientMessage {
        /// Execute a request
        #[serde(rename = "execute")]
        Execute { id: String, request: AgentRequest },
        /// Subscribe to a channel
        #[serde(rename = "subscribe")]
        Subscribe { channel: String },
        /// Unsubscribe from a channel
        #[serde(rename = "unsubscribe")]
        Unsubscribe { channel: String },
        /// Ping for keepalive
        #[serde(rename = "ping")]
        Ping { id: Option<String> },
        /// Register as an agent
        #[serde(rename = "register")]
        Register {
            agent_id: String,
            capabilities: Vec<String>,
        },
        /// Send message to another agent
        #[serde(rename = "agent_message")]
        AgentMessage { to: String, payload: JsonValue },
        /// Broadcast to all agents
        #[serde(rename = "broadcast")]
        Broadcast { channel: String, payload: JsonValue },
    }

    /// WebSocket message to client
    #[derive(Debug, Clone, serde::Serialize)]
    #[serde(tag = "type")]
    pub enum WsServerMessage {
        /// Response to an execute request
        #[serde(rename = "response")]
        Response { id: String, response: AgentResponse },
        /// Streaming event (progress, data, etc.)
        #[serde(rename = "stream")]
        Stream { id: String, event: StreamEvent },
        /// Channel message (from subscription)
        #[serde(rename = "channel")]
        Channel { channel: String, payload: JsonValue },
        /// Pong response
        #[serde(rename = "pong")]
        Pong { id: Option<String>, timestamp: u64 },
        /// Error message
        #[serde(rename = "error")]
        Error { id: Option<String>, message: String },
        /// Agent message received
        #[serde(rename = "agent_message")]
        AgentMessage { from: String, payload: JsonValue },
        /// Agent registered confirmation
        #[serde(rename = "registered")]
        Registered { agent_id: String },
        /// List of connected agents
        #[serde(rename = "agents")]
        Agents { agents: Vec<AgentInfo> },
    }

    /// Information about a connected agent
    #[derive(Debug, Clone, serde::Serialize)]
    pub struct AgentInfo {
        pub id: String,
        pub capabilities: Vec<String>,
        pub connected_at: u64,
    }

    // ========================================================================
    // Agent Orchestration State
    // ========================================================================

    /// Shared state for agent orchestration
    #[derive(Debug)]
    pub struct OrchestratorState {
        /// Connected agents by ID
        agents: RwLock<HashMap<String, AgentConnection>>,
        /// Broadcast channels for pub/sub
        channels: RwLock<HashMap<String, broadcast::Sender<JsonValue>>>,
        /// Task queue for distributed work
        task_queue: RwLock<Vec<Task>>,
        /// Workflow definitions
        workflows: RwLock<HashMap<String, Workflow>>,
        /// Marketplace registry client
        marketplace: RwLock<Option<RegistryClient>>,
    }

    #[derive(Debug)]
    struct AgentConnection {
        info: AgentInfo,
        sender: mpsc::Sender<WsServerMessage>,
    }

    /// A task in the orchestration queue
    #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
    pub struct Task {
        pub id: String,
        pub name: String,
        pub payload: JsonValue,
        pub status: TaskStatus,
        pub assigned_to: Option<String>,
        pub created_at: u64,
        pub priority: i32,
    }

    #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
    pub enum TaskStatus {
        Pending,
        Assigned,
        Running,
        Completed,
        Failed,
    }

    /// A workflow definition
    #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
    pub struct Workflow {
        pub id: String,
        pub name: String,
        pub steps: Vec<WorkflowStep>,
        pub current_step: usize,
        pub status: WorkflowStatus,
        pub context: JsonValue,
    }

    #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
    pub struct WorkflowStep {
        pub name: String,
        pub agent_capability: Option<String>,
        pub request: AgentRequest,
        pub on_success: Option<String>,
        pub on_failure: Option<String>,
    }

    #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
    pub enum WorkflowStatus {
        Pending,
        Running,
        Paused,
        Completed,
        Failed,
    }

    impl OrchestratorState {
        pub fn new() -> Self {
            let marketplace_client = RegistryClient::new(None).ok();
            Self {
                agents: RwLock::new(HashMap::new()),
                channels: RwLock::new(HashMap::new()),
                task_queue: RwLock::new(Vec::new()),
                workflows: RwLock::new(HashMap::new()),
                marketplace: RwLock::new(marketplace_client),
            }
        }

        pub async fn register_agent(
            &self,
            agent_id: String,
            capabilities: Vec<String>,
            sender: mpsc::Sender<WsServerMessage>,
        ) {
            let info = AgentInfo {
                id: agent_id.clone(),
                capabilities,
                connected_at: std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_secs(),
            };
            let conn = AgentConnection { info, sender };
            self.agents.write().await.insert(agent_id, conn);
        }

        pub async fn unregister_agent(&self, agent_id: &str) {
            self.agents.write().await.remove(agent_id);
        }

        pub async fn get_agents(&self) -> Vec<AgentInfo> {
            self.agents
                .read()
                .await
                .values()
                .map(|c| c.info.clone())
                .collect()
        }

        pub async fn send_to_agent(&self, agent_id: &str, message: WsServerMessage) -> bool {
            if let Some(conn) = self.agents.read().await.get(agent_id) {
                conn.sender.send(message).await.is_ok()
            } else {
                false
            }
        }

        pub async fn broadcast_to_channel(&self, channel: &str, payload: JsonValue) {
            let channels = self.channels.read().await;
            if let Some(tx) = channels.get(channel) {
                // `send` fails only when no receiver is left, so the entry is
                // dead weight: release the read lock and prune it.
                if tx.send(payload).is_err() {
                    drop(channels);
                    self.prune_idle_channels().await;
                }
            }
        }

        /// Remove every channel that has no live subscriber.
        ///
        /// Safe against a concurrent subscribe: `subscribe_to_channel` also
        /// takes the write lock, so a channel is either pruned before a new
        /// subscriber arrives (which then creates a fresh sender) or it has
        /// that subscriber by the time it is checked.
        pub async fn prune_idle_channels(&self) {
            self.channels
                .write()
                .await
                .retain(|_, tx| tx.receiver_count() > 0);
        }

        pub async fn subscribe_to_channel(&self, channel: &str) -> broadcast::Receiver<JsonValue> {
            let mut channels = self.channels.write().await;
            if let Some(tx) = channels.get(channel) {
                tx.subscribe()
            } else {
                let (tx, rx) = broadcast::channel(100);
                channels.insert(channel.to_string(), tx);
                rx
            }
        }

        pub async fn add_task(&self, task: Task) {
            self.task_queue.write().await.push(task);
        }

        pub async fn get_pending_tasks(&self) -> Vec<Task> {
            self.task_queue
                .read()
                .await
                .iter()
                .filter(|t| t.status == TaskStatus::Pending)
                .cloned()
                .collect()
        }

        pub async fn claim_task(&self, task_id: &str, agent_id: &str) -> Option<Task> {
            let mut queue = self.task_queue.write().await;
            if let Some(task) = queue
                .iter_mut()
                .find(|t| t.id == task_id && t.status == TaskStatus::Pending)
            {
                task.status = TaskStatus::Assigned;
                task.assigned_to = Some(agent_id.to_string());
                Some(task.clone())
            } else {
                None
            }
        }

        pub async fn complete_task(&self, task_id: &str, success: bool) {
            let mut queue = self.task_queue.write().await;
            if let Some(task) = queue.iter_mut().find(|t| t.id == task_id) {
                task.status = if success {
                    TaskStatus::Completed
                } else {
                    TaskStatus::Failed
                };
            }
        }

        pub async fn add_workflow(&self, workflow: Workflow) {
            self.workflows
                .write()
                .await
                .insert(workflow.id.clone(), workflow);
        }

        pub async fn get_workflow(&self, workflow_id: &str) -> Option<Workflow> {
            self.workflows.read().await.get(workflow_id).cloned()
        }
    }

    impl Default for OrchestratorState {
        fn default() -> Self {
            Self::new()
        }
    }

    /// Server-Sent Event for streaming responses
    #[derive(Debug, Clone, serde::Serialize)]
    pub struct StreamEvent {
        /// Event type: "start", "progress", "data", "complete", "error"
        pub event: String,
        /// Event data (JSON value)
        pub data: JsonValue,
        /// Optional event ID for reconnection
        #[serde(skip_serializing_if = "Option::is_none")]
        pub id: Option<String>,
    }

    impl StreamEvent {
        pub fn start(message: &str) -> Self {
            Self {
                event: "start".to_string(),
                data: json!({ "message": message }),
                id: None,
            }
        }

        pub fn progress(current: usize, total: usize, message: &str) -> Self {
            Self {
                event: "progress".to_string(),
                data: json!({
                    "current": current,
                    "total": total,
                    "percentage": if total > 0 { (current as f64 / total as f64) * 100.0 } else { 0.0 },
                    "message": message
                }),
                id: None,
            }
        }

        pub fn data(result: JsonValue, result_type: Option<&str>) -> Self {
            Self {
                event: "data".to_string(),
                data: json!({
                    "result": result,
                    "result_type": result_type
                }),
                id: None,
            }
        }

        pub fn complete(result: JsonValue, result_type: Option<&str>) -> Self {
            Self {
                event: "complete".to_string(),
                data: json!({
                    "success": true,
                    "result": result,
                    "result_type": result_type
                }),
                id: None,
            }
        }

        pub fn error(message: &str) -> Self {
            Self {
                event: "error".to_string(),
                data: json!({
                    "success": false,
                    "error": message
                }),
                id: None,
            }
        }

        /// A chunk of an array/table result, streamed so clients consume rows
        /// incrementally and can early-stop without buffering the whole result.
        pub fn chunk(seq: usize, rows: &[JsonValue], total: usize) -> Self {
            Self {
                event: "chunk".to_string(),
                data: json!({ "seq": seq, "rows": rows, "total": total }),
                id: None,
            }
        }

        /// Format as SSE text
        pub fn to_sse(&self) -> String {
            let mut output = String::new();
            output.push_str(&format!("event: {}\n", self.event));
            if let Some(ref id) = self.id {
                output.push_str(&format!("id: {}\n", id));
            }
            output.push_str(&format!(
                "data: {}\n\n",
                serde_json::to_string(&self.data).unwrap_or_default()
            ));
            output
        }
    }

    /// Agent API server configuration
    #[derive(Debug, Clone)]
    pub struct AgentApiConfig {
        pub host: String,
        pub port: u16,
        pub enable_cors: bool,
        /// Bearer token required on every route except `/health`.
        ///
        /// `None` means "generate one at startup and print it", which is the
        /// safe default. It is deliberately not possible to configure the server
        /// with no authentication at all: `/api/v1/eval` evaluates arbitrary
        /// IronShell code, so an unauthenticated listener is remote code
        /// execution for anyone who can reach the port — including, when CORS is
        /// enabled, any web page the user happens to visit while the server is
        /// running.
        pub auth_token: Option<String>,
        /// Deadline for a single request/response route, in seconds.
        ///
        /// Without one, an authenticated caller can hold a worker indefinitely
        /// — `/api/v1/eval` evaluates arbitrary code, so an infinite loop is a
        /// one-line request. Enough of those and the server stops answering.
        ///
        /// This does *not* apply to the SSE `/api/v1/stream/*` routes or the
        /// WebSocket, which are long-lived by design; see `build_router`.
        ///
        /// `0` disables the deadline, which reintroduces the exhaustion above.
        /// It exists because a legitimately long evaluation is a real use case,
        /// but prefer raising the value over disabling it.
        pub request_timeout_secs: u64,
    }

    impl Default for AgentApiConfig {
        fn default() -> Self {
            Self {
                host: "127.0.0.1".to_string(),
                port: 3002,
                enable_cors: true,
                auth_token: None,
                // Generous enough that ordinary evaluation is unaffected, short
                // enough that a wedged request frees its worker the same minute.
                request_timeout_secs: 300,
            }
        }
    }

    /// Mint a bearer token, matching `auth::generate_token`'s construction.
    fn generate_api_token() -> Result<String> {
        use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
        let mut bytes = [0u8; 32];
        crate::safety::ic_fill_random("agent_api", &mut bytes)?;
        Ok(URL_SAFE_NO_PAD.encode(bytes))
    }

    /// Constant-time comparison, so a token cannot be recovered a byte at a time
    /// by timing the response.
    fn tokens_match(presented: &str, expected: &str) -> bool {
        let (a, b) = (presented.as_bytes(), expected.as_bytes());
        if a.len() != b.len() {
            return false;
        }
        let mut diff = 0u8;
        for (x, y) in a.iter().zip(b.iter()) {
            diff |= x ^ y;
        }
        diff == 0
    }

    /// Start the Agent API HTTP server
    pub async fn start_agent_api_server(config: AgentApiConfig) -> Result<()> {
        // Create shared orchestrator state
        let state = Arc::new(OrchestratorState::new());

        // Long-lived routes, kept out of the timed router below. An SSE stream
        // and a WebSocket are *supposed* to stay open, so a per-request deadline
        // would sever them at the deadline rather than protect anything.
        let long_lived = Router::new()
            // Streaming endpoints (SSE)
            .route("/api/v1/stream/execute", post(handle_stream_execute))
            .route("/api/v1/stream/pipeline", post(handle_stream_pipeline))
            .route("/api/v1/stream/eval", post(handle_stream_eval))
            // WebSocket endpoint for real-time bidirectional communication
            .route(
                "/api/v1/ws",
                get({
                    let state = Arc::clone(&state);
                    move |ws| handle_websocket(ws, state)
                }),
            );

        let mut app = Router::new()
            // Main execution endpoint
            .route("/api/v1/execute", post(handle_execute))
            // Convenience endpoints
            .route("/api/v1/call/:builtin", post(handle_call))
            .route("/api/v1/pipeline", post(handle_pipeline))
            .route("/api/v1/eval", post(handle_eval))
            // Orchestration endpoints
            .route(
                "/api/v1/orchestration/agents",
                get({
                    let state = Arc::clone(&state);
                    move || handle_list_agents(state)
                }),
            )
            .route(
                "/api/v1/orchestration/tasks",
                get({
                    let state = Arc::clone(&state);
                    move || handle_list_tasks(state)
                }),
            )
            .route(
                "/api/v1/orchestration/tasks",
                post({
                    let state = Arc::clone(&state);
                    move |body| handle_create_task(body, state)
                }),
            )
            .route(
                "/api/v1/orchestration/workflows",
                post({
                    let state = Arc::clone(&state);
                    move |body| handle_create_workflow(body, state)
                }),
            )
            .route(
                "/api/v1/orchestration/workflows/:id",
                get({
                    let state = Arc::clone(&state);
                    move |path| handle_get_workflow(path, state)
                }),
            )
            .route(
                "/api/v1/orchestration/workflows/:id/cancel",
                post({
                    let state = Arc::clone(&state);
                    move |path| handle_cancel_workflow(path, state)
                }),
            )
            .route(
                "/api/v1/orchestration/workflows",
                get({
                    let state = Arc::clone(&state);
                    move || handle_list_workflows(state)
                }),
            )
            .route(
                "/api/v1/orchestration/metrics",
                get({
                    let state = Arc::clone(&state);
                    move || handle_orchestration_metrics(state)
                }),
            )
            // Marketplace endpoints
            .route(
                "/api/v1/marketplace/publish",
                post({
                    let state = Arc::clone(&state);
                    move |body| handle_publish_agent(body, state)
                }),
            )
            .route(
                "/api/v1/marketplace/search",
                get({
                    let state = Arc::clone(&state);
                    move |params| handle_marketplace_search(params, state)
                }),
            )
            .route(
                "/api/v1/marketplace/agents",
                get({
                    let state = Arc::clone(&state);
                    move || handle_marketplace_list(state)
                }),
            )
            .route(
                "/api/v1/marketplace/install",
                post({
                    let state = Arc::clone(&state);
                    move |body| handle_marketplace_install(body, state)
                }),
            )
            .route(
                "/api/v1/marketplace/uninstall",
                post({
                    let state = Arc::clone(&state);
                    move |body| handle_marketplace_uninstall(body, state)
                }),
            )
            // Discovery endpoints
            .route("/api/v1/schema", get(handle_schema))
            .route("/api/v1/schema/:format", get(handle_schema_format))
            .route("/api/v1/builtins", get(handle_list_builtins))
            .route("/api/v1/builtins/:name", get(handle_describe_builtin))
            .route("/api/v1/types", get(handle_types))
            // Health check
            .route("/health", get(handle_health));

        // Give the interpreter a slightly tighter budget than the HTTP layer,
        // so evaluation stops itself first and the caller gets an explanation
        // rather than a bare 408 over a thread that is still running. The
        // `max(1)` keeps a 1-second configured timeout from rounding down to
        // zero, which reads as "no deadline" and would silently disable this.
        EVAL_DEADLINE_SECS.store(
            match config.request_timeout_secs {
                0 => 0,
                secs => (secs * 9 / 10).max(1),
            },
            std::sync::atomic::Ordering::Relaxed,
        );

        // The deadline goes on before the merge, so it wraps only the routes
        // added above — `Router::layer` applies to what is already registered.
        if config.request_timeout_secs > 0 {
            app = app.layer(TimeoutLayer::new(Duration::from_secs(
                config.request_timeout_secs,
            )));
        }
        let mut app = app.merge(long_lived);

        // One semaphore shared by every route (`Router::layer` wraps each route
        // separately, so a per-route limit would not be global). Added after the
        // merge so it covers the streaming and WebSocket routes too; it sits
        // inside the auth layer below, so unauthenticated requests never wait
        // for a permit.
        let permits = Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_REQUESTS));
        app = app.layer(axum::middleware::from_fn(move |req, next| {
            admit(
                permits.clone(),
                MAX_CONCURRENT_REQUESTS,
                MAX_QUEUE_WAIT,
                req,
                next,
            )
        }));

        // Security headers. These servers return JSON, never a page, so the
        // policy is simply "this is not a document": no sniffing, no framing,
        // no subresources, no referrer. `no-store` because responses routinely
        // carry file contents and command output that should not sit in an
        // intermediary cache.
        app = app.layer(axum::middleware::from_fn(
            |req: axum::extract::Request, next: axum::middleware::Next| async move {
                let mut res = next.run(req).await;
                let h = res.headers_mut();
                h.insert(
                    "x-content-type-options",
                    header::HeaderValue::from_static("nosniff"),
                );
                h.insert("x-frame-options", header::HeaderValue::from_static("DENY"));
                h.insert(
                    "content-security-policy",
                    header::HeaderValue::from_static("default-src 'none'; frame-ancestors 'none'"),
                );
                h.insert(
                    "referrer-policy",
                    header::HeaderValue::from_static("no-referrer"),
                );
                h.insert(
                    "cache-control",
                    header::HeaderValue::from_static("no-store"),
                );
                res
            },
        ));

        // Authentication. Applied before CORS below so that it wraps every
        // route; `/health` is exempted inside the middleware so liveness probes
        // keep working without a credential.
        let token = Arc::new(match config.auth_token.clone() {
            Some(t) if !t.trim().is_empty() => t,
            _ => generate_api_token()?,
        });
        let auth_token = Arc::clone(&token);
        app = app.layer(axum::middleware::from_fn(
            move |req: axum::extract::Request, next: axum::middleware::Next| {
                let expected = Arc::clone(&auth_token);
                async move {
                    if req.uri().path() == "/health" {
                        return next.run(req).await;
                    }
                    let presented = req
                        .headers()
                        .get(header::AUTHORIZATION)
                        .and_then(|v| v.to_str().ok())
                        .and_then(|v| v.strip_prefix("Bearer "))
                        .map(str::trim)
                        .unwrap_or("");

                    if tokens_match(presented, &expected) {
                        next.run(req).await
                    } else {
                        (
                            StatusCode::UNAUTHORIZED,
                            [(header::WWW_AUTHENTICATE, "Bearer")],
                            Json(serde_json::json!({
                                "error": "unauthorized",
                                "detail": "Send `Authorization: Bearer <token>`. \
                                           The token is printed when the server starts.",
                            })),
                        )
                            .into_response()
                    }
                }
            },
        ));

        if config.enable_cors {
            // `allow_origin(Any)` tells every browser that any web page may talk
            // to this server. That is only tolerable because the bearer token
            // above is required and a cross-origin page cannot read it: without
            // authentication this combination let any site the user visited POST
            // to /api/v1/eval on their loopback interface and execute code.
            app = app.layer(
                CorsLayer::new()
                    .allow_origin(Any)
                    .allow_methods(Any)
                    .allow_headers(Any),
            );
        }

        // `SocketAddr` parses literal addresses only — `localhost:3002` is a
        // perfectly reasonable thing to put in a config file and does *not*
        // parse. Report it instead of panicking the server at startup.
        let bind = format!("{}:{}", config.host, config.port);
        let addr: std::net::SocketAddr = bind.parse().map_err(|e| {
            anyhow::anyhow!(
                "invalid bind address {bind:?}: {e}. \
                 Use a literal IP such as 127.0.0.1 or 0.0.0.0, not a hostname."
            )
        })?;

        if !addr.ip().is_loopback() {
            println!(
                "⚠  Binding {} — this API evaluates arbitrary code, so it is now\n\
                 ⚠  reachable by anything that can route to this host. The bearer\n\
                 ⚠  token below is the only thing standing in front of it.",
                addr.ip()
            );
        }

        println!("🤖 IronShell Agent API starting on http://{}", addr);
        if config.auth_token.is_none() {
            println!("   Auth token (generated): {}", token);
        } else {
            println!("   Auth token: (from configuration)");
        }
        println!("   Send it as: Authorization: Bearer <token>");
        // The safety posture is stated, never assumed. This endpoint evaluates
        // arbitrary code, so which mode it is in is the single most important
        // thing an operator needs to read off the banner.
        match crate::safety::current_mode() {
            crate::safety::Mode::Agent => {
                println!("   Safety: agent mode — dangerous effect classes default-denied");
                println!(
                    "           Writes confined to {}",
                    crate::safety::workspace_root().display()
                );
            }
            other => {
                println!(
                    "⚠  Safety: {other:?} mode (IRONSHELL_MODE set explicitly) — the effect
                     ⚠  gate and the workspace jail are OFF for a surface that evaluates
                     ⚠  arbitrary code. Unset IRONSHELL_MODE to get the agent profile."
                );
            }
        }
        println!("   Supports: OpenAI, Claude, Gemini, Llama, Mistral, Cohere, Grok, DeepSeek,");
        println!("             Bedrock, Azure, Qwen, Ollama, vLLM, HuggingFace, OpenRouter,");
        println!("             Kimi, Yi, GLM, Reka, AI21, Perplexity, Together, Groq, Fireworks");
        println!();
        println!("Endpoints:");
        println!("  POST /api/v1/execute          - Execute any request");
        println!("  POST /api/v1/call/:builtin    - Call a single builtin");
        println!("  POST /api/v1/pipeline         - Execute a pipeline");
        println!("  POST /api/v1/eval             - Evaluate raw code");
        println!("  POST /api/v1/stream/execute   - Stream execution (SSE)");
        println!("  POST /api/v1/stream/pipeline  - Stream pipeline (SSE)");
        println!("  POST /api/v1/stream/eval      - Stream eval (SSE)");
        println!("  GET  /api/v1/ws               - WebSocket (real-time bidirectional)");
        println!("  GET  /api/v1/orchestration/*  - Agent orchestration APIs");
        println!("  POST /api/v1/marketplace/*    - Agent marketplace APIs");
        println!("  GET  /api/v1/schema           - Get language ontology");
        println!("  GET  /api/v1/schema/:format   - Get schema for AI provider");
        println!("  GET  /api/v1/builtins         - List all builtins");
        println!("  GET  /api/v1/builtins/:name   - Describe a builtin");
        println!("  GET  /api/v1/types            - Get type information");

        let listener = tokio::net::TcpListener::bind(addr).await?;
        crate::serve::serve_until(
            listener,
            app,
            crate::serve::shutdown_signal(),
            crate::serve::DRAIN,
        )
        .await?;

        Ok(())
    }

    /// Evaluation deadline for request handlers, in seconds; 0 means none.
    ///
    /// A process-level value rather than router state because these routes are
    /// registered as bare functions and there is one server per process. Set
    /// once in `start_agent_api_server`, read per request.
    static EVAL_DEADLINE_SECS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

    /// The evaluation deadline for the running server, if any.
    ///
    /// Deliberately shorter than the HTTP timeout: the interpreter should stop
    /// on its own *before* the connection is torn out from under it, so the
    /// caller gets a real error explaining what happened rather than a bare
    /// 408 and a thread that keeps working.
    fn eval_deadline() -> Option<std::time::Duration> {
        match EVAL_DEADLINE_SECS.load(std::sync::atomic::Ordering::Relaxed) {
            0 => None,
            secs => Some(std::time::Duration::from_secs(secs)),
        }
    }

    /// Run a request on the blocking pool and shape the response.
    ///
    /// `process_request` is synchronous and can run arbitrary evaluated code,
    /// so calling it directly from an `async fn` pins an async worker for the
    /// whole evaluation. That is not merely a throughput concern: it also made
    /// `TimeoutLayer` decorative, because tower races the deadline against the
    /// inner future *within the same poll*. If the inner call never yields, the
    /// timeout branch is never reached and the deadline cannot fire — measured,
    /// not assumed: before this change a 1-second deadline did not interrupt a
    /// 20-second request at all.
    ///
    /// What this bounds and what it does not. Moving the work here lets the
    /// deadline fire, frees the async worker, and returns 408 to the caller.
    /// Dropping a `spawn_blocking` handle does not cancel the closure, so the
    /// HTTP-level timeout alone would leave the evaluation running on a
    /// blocking-pool thread until it returned by itself.
    ///
    /// `safety::enter_deadline` closes most of that: the interpreter checks it
    /// in `eval_expr`, so runaway recursion and large computations stop rather
    /// than leak a thread. The guard restores the previous value on drop, which
    /// matters here specifically — these threads are pooled and reused, so a
    /// deadline left set would make the *next* request on that thread fail
    /// immediately.
    ///
    /// It still cannot interrupt a builtin already blocked in a syscall
    /// (`sleep`, a subprocess wait, a network read), because those never return
    /// to the interpreter to be asked. That is the remaining gap, and it is
    /// narrower than before rather than closed.
    async fn run_blocking(
        request: AgentRequest,
        deadline: Option<std::time::Duration>,
    ) -> (StatusCode, Json<AgentResponse>) {
        match evaluate_off_thread(request, deadline).await {
            Ok(response) => {
                if response.success {
                    (StatusCode::OK, Json(response))
                } else {
                    (StatusCode::BAD_REQUEST, Json(response))
                }
            }
            Err(error) => (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(handler_failure(error)),
            ),
        }
    }

    /// Evaluate a request on the blocking pool under the evaluation deadline.
    ///
    /// Used by every path that runs `process_request` on behalf of a client,
    /// so none of them pins an async worker for the length of an evaluation.
    /// `Err` means the evaluation panicked; the message is the reason.
    async fn evaluate_off_thread(
        request: AgentRequest,
        deadline: Option<std::time::Duration>,
    ) -> Result<AgentResponse, String> {
        tokio::task::spawn_blocking(move || {
            let _guard = deadline.map(crate::safety::enter_deadline);
            process_request(&request)
        })
        .await
        .map_err(|e| e.to_string())
    }

    /// The response for a handler that failed before producing one. Reported
    /// as data rather than propagated, so one bad request cannot take down the
    /// listener or the connection.
    fn handler_failure(error: String) -> AgentResponse {
        AgentResponse {
            code: None,
            hint: None,
            success: false,
            result: None,
            error: Some(format!("request handler failed: {error}")),
            result_type: None,
            metadata: None,
        }
    }

    // The body is read as text and parsed by `parse_request`, as stdin is, so
    // HTTP refuses what `ish agent execute` refuses (a repeated top-level key)
    // with the same coded error, instead of serde's last-key-wins.
    async fn handle_execute(body: String) -> (StatusCode, Json<AgentResponse>) {
        match crate::agent_api::parse_request(&body) {
            Ok(request) => run_blocking(request, eval_deadline()).await,
            Err(refusal) => (StatusCode::BAD_REQUEST, Json(refusal)),
        }
    }

    async fn handle_call(
        axum::extract::Path(builtin): axum::extract::Path<String>,
        Json(args): Json<JsonValue>,
    ) -> impl IntoResponse {
        // An unknown builtin is a missing resource, not a malformed request.
        if !builtin.contains('.') {
            if let Some(refusal) = crate::agent_api::unknown_builtin_refusal(&builtin) {
                return (StatusCode::NOT_FOUND, Json(refusal));
            }
        }
        run_blocking(AgentRequest::Call { builtin, args }, eval_deadline()).await
    }

    async fn handle_pipeline(Json(body): Json<JsonValue>) -> impl IntoResponse {
        let steps: Vec<PipelineStep> = serde_json::from_value(
            body.get("steps")
                .cloned()
                .unwrap_or(JsonValue::Array(vec![])),
        )
        .unwrap_or_default();

        let input = body.get("input").cloned();

        run_blocking(AgentRequest::Pipeline { steps, input }, eval_deadline()).await
    }

    async fn handle_eval(Json(body): Json<JsonValue>) -> impl IntoResponse {
        let code = match crate::agent_api::eval_code_from_body(&body) {
            Ok(code) => code,
            Err(refusal) => return (StatusCode::BAD_REQUEST, Json(refusal)),
        };

        run_blocking(AgentRequest::Eval { code }, eval_deadline()).await
    }

    // ========================================================================
    // Streaming Handlers (Server-Sent Events)
    // ========================================================================

    /// Stream execution of a generic request
    async fn handle_stream_execute(body: String) -> Response {
        let request = match crate::agent_api::parse_request(&body) {
            Ok(request) => request,
            Err(refusal) => return (StatusCode::BAD_REQUEST, Json(refusal)).into_response(),
        };
        let Some(slot) = try_take_stream_slot(&stream_slots()) else {
            return stream_busy_response();
        };
        let deadline = eval_deadline();
        create_sse_response(async move {
            // Held until the evaluation finishes; dropped with the stream if the client leaves.
            let _slot = slot;
            match evaluate_off_thread(request, deadline).await {
                Ok(response) => stream_events_from_response(response, 50),
                Err(error) => vec![StreamEvent::error(&format!(
                    "request handler failed: {error}"
                ))],
            }
        })
        .into_response()
    }

    /// Build streaming events from a response. A large array result is split into
    /// ordered `chunk` events (default 50 rows) so a client consumes incrementally
    /// and can early-stop instead of buffering the whole result; smaller results
    /// and non-arrays emit a single `complete` event; failures emit `error`.
    pub fn stream_events_from_response(
        response: AgentResponse,
        chunk_rows: usize,
    ) -> Vec<StreamEvent> {
        let mut events = vec![StreamEvent::start("Processing request...")];
        if !response.success {
            events.push(StreamEvent::error(
                &response
                    .error
                    .unwrap_or_else(|| "Unknown error".to_string()),
            ));
            return events;
        }
        let result = response.result.unwrap_or(JsonValue::Null);
        match &result {
            JsonValue::Array(rows) if rows.len() > chunk_rows.max(1) => {
                let n = chunk_rows.max(1);
                let total = rows.len();
                for (seq, chunk) in rows.chunks(n).enumerate() {
                    events.push(StreamEvent::chunk(seq, chunk, total));
                }
                events.push(StreamEvent::complete(
                    json!({ "streamed_rows": total }),
                    response.result_type.as_deref(),
                ));
            }
            _ => events.push(StreamEvent::complete(
                result,
                response.result_type.as_deref(),
            )),
        }
        events
    }

    /// The SSE events for a pipeline: `start`, then a `progress` after each step
    /// that really completed, then `complete` (with the same result
    /// `process_request` gives for the pipeline) or `error` naming the step that
    /// failed. `deadline` bounds the evaluation as it does for the other handlers.
    pub fn stream_pipeline_events(
        steps: &[PipelineStep],
        input: Option<&JsonValue>,
        deadline: Option<std::time::Duration>,
    ) -> Vec<StreamEvent> {
        let total = steps.len();
        let mut events = vec![StreamEvent::start(&format!(
            "Starting pipeline with {} steps",
            total
        ))];
        let _guard = deadline.map(crate::safety::enter_deadline);

        let outcome = run_pipeline_steps(steps, input, |i| {
            events.push(StreamEvent::progress(
                i + 1,
                total,
                &format!("Completed step {}/{}: {}", i + 1, total, steps[i].builtin),
            ));
        });
        match outcome {
            Ok((value, code)) => {
                let response = response_from_value(value, &code);
                events.push(StreamEvent::complete(
                    response.result.unwrap_or(JsonValue::Null),
                    response.result_type.as_deref(),
                ));
            }
            Err(failure) => {
                let message = match failure.step {
                    Some(i) => format!(
                        "Step {}/{} ({}) failed: {}",
                        i + 1,
                        total,
                        steps[i].builtin,
                        failure.message
                    ),
                    None => failure.message,
                };
                events.push(StreamEvent::error(&message));
            }
        }
        events
    }

    /// Stream pipeline execution with progress updates
    async fn handle_stream_pipeline(Json(body): Json<JsonValue>) -> Response {
        let Some(slot) = try_take_stream_slot(&stream_slots()) else {
            return stream_busy_response();
        };
        let steps: Vec<PipelineStep> = serde_json::from_value(
            body.get("steps")
                .cloned()
                .unwrap_or(JsonValue::Array(vec![])),
        )
        .unwrap_or_default();

        let input = body.get("input").cloned();
        let deadline = eval_deadline();

        create_sse_response(async move {
            // Held until the evaluation finishes; dropped with the stream if the client leaves.
            let _slot = slot;
            // Evaluation is synchronous: keep it off the async workers, as run_blocking does.
            match tokio::task::spawn_blocking(move || {
                stream_pipeline_events(&steps, input.as_ref(), deadline)
            })
            .await
            {
                Ok(events) => events,
                Err(e) => vec![StreamEvent::error(&format!("request handler failed: {e}"))],
            }
        })
        .into_response()
    }

    /// The SSE events for a streamed evaluation of `code`: `start`, then the
    /// result in `chunk` events of 50 items, then `complete`; or `error`.
    /// `deadline` bounds the evaluation, as it does for the other handlers.
    pub fn stream_eval_events(
        code: &str,
        deadline: Option<std::time::Duration>,
    ) -> Vec<StreamEvent> {
        let mut events = vec![StreamEvent::start("Evaluating code...")];
        let _guard = deadline.map(crate::safety::enter_deadline);

        // Stream-evaluate: results are produced incrementally (element-by-element
        // for a streamable `source | map/where/…` pipeline) rather than the whole
        // value being materialized first, then chunked for the wire.
        let mut env = crate::modules::env_with_modules();
        let mut items: Vec<JsonValue> = Vec::new();
        let started = std::time::Instant::now();
        let result = {
            let mut emit = |v: crate::value::Value| items.push(v.to_json());
            crate::eval::eval_stream(code, &mut env, &mut emit)
        };
        crate::record::note(
            crate::record::Surface::Api,
            code,
            &result,
            started.elapsed(),
        );
        match result {
            Ok(n) => {
                for (seq, chunk) in items.chunks(50).enumerate() {
                    events.push(StreamEvent::chunk(seq, chunk, n));
                }
                events.push(StreamEvent::complete(
                    json!({ "streamed_items": n }),
                    Some("stream"),
                ));
            }
            Err(e) => events.push(StreamEvent::error(&format!("{}", e))),
        }
        events
    }

    /// Stream code evaluation
    async fn handle_stream_eval(Json(body): Json<JsonValue>) -> Response {
        let code = match crate::agent_api::eval_code_from_body(&body) {
            Ok(code) => code,
            Err(refusal) => return (StatusCode::BAD_REQUEST, Json(refusal)).into_response(),
        };
        let Some(slot) = try_take_stream_slot(&stream_slots()) else {
            return stream_busy_response();
        };
        let deadline = eval_deadline();

        create_sse_response(async move {
            // Held until the evaluation finishes; dropped with the stream if the client leaves.
            let _slot = slot;
            // Evaluation is synchronous: keep it off the async workers, as run_blocking does.
            match tokio::task::spawn_blocking(move || stream_eval_events(&code, deadline)).await {
                Ok(events) => events,
                Err(e) => vec![StreamEvent::error(&format!("request handler failed: {e}"))],
            }
        })
        .into_response()
    }

    /// Helper to create SSE response from events
    fn create_sse_response<F>(event_generator: F) -> impl IntoResponse
    where
        F: std::future::Future<Output = Vec<StreamEvent>> + Send + 'static,
    {
        let stream = async_stream::stream! {
            let events = event_generator.await;
            for event in events {
                yield Ok::<_, Infallible>(event.to_sse());
            }
        };

        Response::builder()
            .status(StatusCode::OK)
            .header(header::CONTENT_TYPE, "text/event-stream")
            .header(header::CACHE_CONTROL, "no-cache")
            .header("X-Accel-Buffering", "no")
            .body(Body::from_stream(stream))
            .unwrap()
    }

    async fn handle_schema(
        axum::extract::Query(params): axum::extract::Query<HashMap<String, String>>,
    ) -> impl IntoResponse {
        let request = AgentRequest::Schema {
            format: SchemaFormat::Ontology,
            module: params.get("module").cloned(),
            q: params.get("q").cloned(),
        };
        schema_response(&request)
    }

    async fn handle_schema_format(
        axum::extract::Path(format): axum::extract::Path<String>,
        axum::extract::Query(params): axum::extract::Query<HashMap<String, String>>,
    ) -> impl IntoResponse {
        let Some(schema_format) = parse_schema_format(&format) else {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({
                    "success": false,
                    "error": format!("Unknown schema format: '{format}'. Supported: {SCHEMA_FORMATS_HINT}"),
                    "metadata": { "code": "E_BAD_ARG" },
                })),
            );
        };

        let request = AgentRequest::Schema {
            format: schema_format,
            module: params.get("module").cloned(),
            q: params.get("q").cloned(),
        };
        schema_response(&request)
    }

    /// A schema answer, or 400 when the request's filter is refused.
    fn schema_response(request: &AgentRequest) -> (StatusCode, Json<JsonValue>) {
        let response = process_request(request);
        let status = if response.success {
            StatusCode::OK
        } else {
            StatusCode::BAD_REQUEST
        };
        (status, Json(json!(response)))
    }

    async fn handle_list_builtins(
        axum::extract::Query(params): axum::extract::Query<HashMap<String, String>>,
    ) -> impl IntoResponse {
        let category = params.get("category").cloned();
        let request = AgentRequest::ListBuiltins { category };
        Json(process_request(&request))
    }

    async fn handle_describe_builtin(
        axum::extract::Path(name): axum::extract::Path<String>,
    ) -> impl IntoResponse {
        let request = AgentRequest::Describe { builtin: name };
        let response = process_request(&request);
        if response.success {
            (StatusCode::OK, Json(response))
        } else {
            (StatusCode::NOT_FOUND, Json(response))
        }
    }

    async fn handle_types() -> impl IntoResponse {
        let request = AgentRequest::TypeInfo { type_name: None };
        Json(process_request(&request))
    }

    async fn handle_health() -> impl IntoResponse {
        Json(json!({
            "status": "healthy",
            "service": "ironshell-agent-api",
            "version": env!("CARGO_PKG_VERSION"),
            "features": {
                "websocket": true,
                "sse_streaming": true,
                "orchestration": true
            },
            "supported_agents": [
                "openai", "azure_openai", "claude", "gemini",
                "llama", "mistral", "cohere", "grok", "deepseek",
                "bedrock", "qwen", "ollama", "vllm", "huggingface", "openrouter",
                "kimi", "yi", "glm", "reka", "ai21", "perplexity", "together", "groq", "fireworks"
            ],
            "schema_formats": {
                "openai_family": ["openai", "gpt", "chatgpt", "azure", "azure_openai"],
                "anthropic": ["claude", "anthropic"],
                "google": ["gemini", "google"],
                "meta": ["llama", "meta", "llama3"],
                "mistral": ["mistral", "codestral", "pixtral"],
                "cohere": ["cohere", "command", "command-r"],
                "xai": ["grok", "xai"],
                "deepseek": ["deepseek", "deepseek-r1"],
                "cloud": ["bedrock", "aws", "amazon"],
                "alibaba": ["qwen", "alibaba", "dashscope"],
                "chinese_ai": ["kimi", "moonshot", "yi", "01ai", "glm", "chatglm", "zhipu"],
                "inference_platforms": ["together", "together-ai", "groq", "fireworks", "fireworks-ai"],
                "additional_sota": ["reka", "ai21", "jamba", "perplexity", "sonar"],
                "local": ["ollama", "lmstudio", "vllm"],
                "platforms": ["huggingface", "hf", "tgi", "openrouter"],
                "standard": ["json", "jsonschema", "ontology", "compact"]
            }
        }))
    }

    // ========================================================================
    // WebSocket Handler
    // ========================================================================

    /// Copy one channel's messages into a connection's outbound queue.
    ///
    /// A lagging receiver (the publisher outran its buffer) skips the messages
    /// it missed and keeps going; only a closed channel or a closed connection
    /// ends the forwarder.
    async fn forward_channel(
        mut rx: broadcast::Receiver<JsonValue>,
        out: mpsc::Sender<WsServerMessage>,
        channel: String,
    ) {
        loop {
            match rx.recv().await {
                Ok(payload) => {
                    let msg = WsServerMessage::Channel {
                        channel: channel.clone(),
                        payload,
                    };
                    if out.send(msg).await.is_err() {
                        break;
                    }
                }
                Err(broadcast::error::RecvError::Lagged(_)) => continue,
                Err(broadcast::error::RecvError::Closed) => break,
            }
        }
    }

    /// The forwarding tasks for one WebSocket connection's channel
    /// subscriptions, keyed by channel name.
    ///
    /// A spawned task keeps running after its handle is dropped, so each one
    /// is aborted explicitly: on Unsubscribe, when the same channel is
    /// subscribed again, and when the connection closes. Dropping the set
    /// aborts whatever is left, as a backstop.
    #[derive(Default)]
    struct SubscriptionTasks {
        tasks: HashMap<String, JoinHandle<()>>,
    }

    impl SubscriptionTasks {
        fn insert(&mut self, channel: String, task: JoinHandle<()>) {
            if let Some(previous) = self.tasks.insert(channel, task) {
                previous.abort();
            }
        }

        /// Stop the subscription to `channel`. Returns whether one existed.
        async fn remove(&mut self, channel: &str) -> bool {
            match self.tasks.remove(channel) {
                Some(task) => {
                    stop_task(task).await;
                    true
                }
                None => false,
            }
        }

        /// Stop every subscription. Used when the connection closes.
        async fn stop_all(&mut self) {
            let tasks: Vec<JoinHandle<()>> = self.tasks.drain().map(|(_, t)| t).collect();
            for task in tasks {
                stop_task(task).await;
            }
        }

        #[cfg(test)]
        fn len(&self) -> usize {
            self.tasks.len()
        }
    }

    impl Drop for SubscriptionTasks {
        fn drop(&mut self) {
            for (_, task) in self.tasks.drain() {
                task.abort();
            }
        }
    }

    /// Abort a task and wait for it to finish. Waiting matters: the task owns
    /// a broadcast receiver, and that receiver only goes away once the future
    /// is dropped, which is when the task actually stops. Waiting lets the
    /// channel pruning that follows see the subscriber gone.
    async fn stop_task(task: JoinHandle<()>) {
        task.abort();
        let _ = task.await;
    }

    async fn handle_websocket(
        ws: WebSocketUpgrade,
        state: Arc<OrchestratorState>,
    ) -> impl IntoResponse {
        ws.on_upgrade(move |socket| handle_websocket_connection(socket, state))
    }

    async fn handle_websocket_connection(socket: WebSocket, state: Arc<OrchestratorState>) {
        use futures_util::{SinkExt, StreamExt};

        let (mut sender, mut receiver) = socket.split();
        let (tx, mut rx) = mpsc::channel::<WsServerMessage>(100);
        let mut agent_id: Option<String> = None;
        let mut subscriptions = SubscriptionTasks::default();

        // Task for sending messages to the client
        let send_task = tokio::spawn(async move {
            while let Some(msg) = rx.recv().await {
                if let Ok(json) = serde_json::to_string(&msg) {
                    if sender.send(Message::Text(json)).await.is_err() {
                        break;
                    }
                }
            }
        });

        // Process incoming messages
        while let Some(result) = receiver.next().await {
            match result {
                Ok(Message::Text(text)) => {
                    if let Ok(msg) = serde_json::from_str::<WsClientMessage>(&text) {
                        let response = match msg {
                            WsClientMessage::Execute { id, request } => {
                                // The slot is held for the whole evaluation; a refused
                                // request gets a coded response instead of queueing.
                                let response = match try_take_stream_slot(&stream_slots()) {
                                    Some(_slot) => {
                                        match evaluate_off_thread(request, eval_deadline()).await {
                                            Ok(response) => response,
                                            Err(error) => handler_failure(error),
                                        }
                                    }
                                    None => ws_busy_response(),
                                };
                                Some(WsServerMessage::Response { id, response })
                            }
                            WsClientMessage::Ping { id } => {
                                let timestamp = std::time::SystemTime::now()
                                    .duration_since(std::time::UNIX_EPOCH)
                                    .unwrap()
                                    .as_secs();
                                Some(WsServerMessage::Pong { id, timestamp })
                            }
                            WsClientMessage::Register {
                                agent_id: aid,
                                capabilities,
                            } => {
                                agent_id = Some(aid.clone());
                                state
                                    .register_agent(aid.clone(), capabilities.clone(), tx.clone())
                                    .await;

                                // Broadcast agent connection to subscribers
                                state
                                    .broadcast_to_channel(
                                        "agents",
                                        json!({
                                            "type": "agent_connected",
                                            "agent": {
                                                "id": aid,
                                                "capabilities": capabilities,
                                                "status": "online",
                                                "connectedAt": std::time::SystemTime::now()
                                                    .duration_since(std::time::UNIX_EPOCH)
                                                    .unwrap()
                                                    .as_millis() as u64
                                            }
                                        }),
                                    )
                                    .await;

                                Some(WsServerMessage::Registered { agent_id: aid })
                            }
                            WsClientMessage::AgentMessage { to, payload } => {
                                if let Some(ref from) = agent_id {
                                    let msg = WsServerMessage::AgentMessage {
                                        from: from.clone(),
                                        payload,
                                    };
                                    state.send_to_agent(&to, msg).await;
                                }
                                None
                            }
                            WsClientMessage::Broadcast { channel, payload } => {
                                state.broadcast_to_channel(&channel, payload).await;
                                None
                            }
                            WsClientMessage::Subscribe { channel } => {
                                // Re-subscribing replaces the old forwarder rather than doubling it.
                                subscriptions.remove(&channel).await;
                                let rx = state.subscribe_to_channel(&channel).await;
                                let task =
                                    tokio::spawn(forward_channel(rx, tx.clone(), channel.clone()));
                                subscriptions.insert(channel, task);
                                None
                            }
                            WsClientMessage::Unsubscribe { channel } => {
                                subscriptions.remove(&channel).await;
                                state.prune_idle_channels().await;
                                None
                            }
                        };

                        if let Some(response) = response {
                            if tx.send(response).await.is_err() {
                                break;
                            }
                        }
                    }
                }
                Ok(Message::Close(_)) => break,
                Err(_) => break,
                _ => {}
            }
        }

        // Clean up
        if let Some(ref aid) = agent_id {
            state
                .broadcast_to_channel(
                    "agents",
                    json!({
                        "type": "agent_disconnected",
                        "agentId": aid
                    }),
                )
                .await;
            state.unregister_agent(aid).await;
        }
        subscriptions.stop_all().await;
        state.prune_idle_channels().await;
        send_task.abort();
    }

    // ========================================================================
    // Orchestration Handlers
    // ========================================================================

    async fn handle_list_agents(state: Arc<OrchestratorState>) -> impl IntoResponse {
        let agents = state.get_agents().await;
        Json(json!({
            "success": true,
            "agents": agents
        }))
    }

    async fn handle_list_tasks(state: Arc<OrchestratorState>) -> impl IntoResponse {
        let tasks = state.get_pending_tasks().await;
        Json(json!({
            "success": true,
            "tasks": tasks
        }))
    }

    async fn handle_create_task(
        Json(body): Json<JsonValue>,
        state: Arc<OrchestratorState>,
    ) -> impl IntoResponse {
        let task = Task {
            id: uuid::Uuid::new_v4().to_string(),
            name: body
                .get("name")
                .and_then(|v| v.as_str())
                .unwrap_or("unnamed")
                .to_string(),
            payload: body.get("payload").cloned().unwrap_or(JsonValue::Null),
            status: TaskStatus::Pending,
            assigned_to: None,
            created_at: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_secs(),
            priority: body.get("priority").and_then(|v| v.as_i64()).unwrap_or(0) as i32,
        };
        let task_id = task.id.clone();
        state.add_task(task).await;

        (
            StatusCode::CREATED,
            Json(json!({
                "success": true,
                "task_id": task_id
            })),
        )
    }

    async fn handle_create_workflow(
        Json(body): Json<JsonValue>,
        state: Arc<OrchestratorState>,
    ) -> impl IntoResponse {
        // An absent `steps` is an empty workflow. A present value that does not
        // parse is refused, and nothing is stored.
        let steps: Vec<WorkflowStep> = match body.get("steps") {
            None => Vec::new(),
            Some(value) => match serde_json::from_value(value.clone()) {
                Ok(steps) => steps,
                Err(e) => {
                    return (
                        StatusCode::BAD_REQUEST,
                        Json(json!({
                            "success": false,
                            "error": format!("invalid steps: {}", e)
                        })),
                    );
                }
            },
        };

        let workflow = Workflow {
            id: uuid::Uuid::new_v4().to_string(),
            name: body
                .get("name")
                .and_then(|v| v.as_str())
                .unwrap_or("unnamed")
                .to_string(),
            steps,
            current_step: 0,
            status: WorkflowStatus::Pending,
            context: body
                .get("context")
                .cloned()
                .unwrap_or(JsonValue::Object(Default::default())),
        };
        let workflow_id = workflow.id.clone();
        let workflow_name = workflow.name.clone();
        let total_steps = workflow.steps.len();
        state.add_workflow(workflow).await;

        // Broadcast workflow creation
        state
            .broadcast_to_channel(
                "workflows",
                json!({
                    "type": "workflow_created",
                    "workflow": {
                        "id": workflow_id,
                        "name": workflow_name,
                        "status": "pending",
                        "currentStep": 0,
                        "totalSteps": total_steps,
                        "startedAt": std::time::SystemTime::now()
                            .duration_since(std::time::UNIX_EPOCH)
                            .unwrap()
                            .as_millis() as u64,
                        "steps": []
                    }
                }),
            )
            .await;

        (
            StatusCode::CREATED,
            Json(json!({
                "success": true,
                "workflow_id": workflow_id
            })),
        )
    }

    async fn handle_get_workflow(
        axum::extract::Path(workflow_id): axum::extract::Path<String>,
        state: Arc<OrchestratorState>,
    ) -> impl IntoResponse {
        if let Some(workflow) = state.get_workflow(&workflow_id).await {
            (
                StatusCode::OK,
                Json(json!({
                    "success": true,
                    "workflow": workflow
                })),
            )
        } else {
            (
                StatusCode::NOT_FOUND,
                Json(json!({
                    "success": false,
                    "error": "Workflow not found"
                })),
            )
        }
    }

    async fn handle_cancel_workflow(
        axum::extract::Path(workflow_id): axum::extract::Path<String>,
        state: Arc<OrchestratorState>,
    ) -> impl IntoResponse {
        if let Some(mut workflow) = state.get_workflow(&workflow_id).await {
            workflow.status = WorkflowStatus::Failed;
            state.add_workflow(workflow).await;

            // Broadcast cancellation
            state
                .broadcast_to_channel(
                    "workflows",
                    json!({
                        "type": "workflow_update",
                        "workflow": {
                            "id": workflow_id,
                            "status": "cancelled"
                        }
                    }),
                )
                .await;

            (
                StatusCode::OK,
                Json(json!({
                    "success": true,
                    "message": "Workflow cancelled"
                })),
            )
        } else {
            (
                StatusCode::NOT_FOUND,
                Json(json!({
                    "success": false,
                    "error": "Workflow not found"
                })),
            )
        }
    }

    async fn handle_list_workflows(state: Arc<OrchestratorState>) -> impl IntoResponse {
        let workflows = state.workflows.read().await;
        let list: Vec<_> = workflows.values().cloned().collect();
        Json(json!({
            "success": true,
            "workflows": list
        }))
    }

    async fn handle_orchestration_metrics(state: Arc<OrchestratorState>) -> impl IntoResponse {
        let agents = state.get_agents().await;
        let workflows = state.workflows.read().await;
        let tasks = state.task_queue.read().await;

        let running_workflows = workflows
            .values()
            .filter(|w| w.status == WorkflowStatus::Running)
            .count();
        let completed_workflows = workflows
            .values()
            .filter(|w| w.status == WorkflowStatus::Completed)
            .count();
        let failed_workflows = workflows
            .values()
            .filter(|w| w.status == WorkflowStatus::Failed)
            .count();
        let pending_tasks = tasks
            .iter()
            .filter(|t| t.status == TaskStatus::Pending)
            .count();

        Json(json!({
            "success": true,
            "metrics": {
                "agents": {
                    "total": agents.len(),
                    "online": agents.len(),
                },
                "workflows": {
                    "total": workflows.len(),
                    "running": running_workflows,
                    "completed": completed_workflows,
                    "failed": failed_workflows,
                },
                "tasks": {
                    "total": tasks.len(),
                    "pending": pending_tasks,
                },
                "timestamp": std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_secs()
            }
        }))
    }

    async fn handle_publish_agent(
        Json(body): Json<JsonValue>,
        state: Arc<OrchestratorState>,
    ) -> impl IntoResponse {
        let agent_name = body
            .get("name")
            .and_then(|v| v.as_str())
            .unwrap_or("unnamed")
            .to_string();

        // Announce on the agents channel only. The agent is not stored, and it
        // is not sent to a marketplace registry.
        state
            .broadcast_to_channel(
                "agents",
                json!({
                    "type": "agent_published",
                    "agent": {
                        "name": agent_name,
                        "description": body.get("description"),
                        "systemPrompt": body.get("systemPrompt"),
                        "tools": body.get("tools"),
                        "model": body.get("model"),
                    }
                }),
            )
            .await;

        (
            StatusCode::ACCEPTED,
            Json(json!({
                "success": true,
                "stored": false,
                "message": format!(
                    "Agent '{}' announced to subscribers of the agents channel; it was not stored or sent to a marketplace registry",
                    agent_name
                )
            })),
        )
    }

    async fn handle_marketplace_search(
        axum::extract::Query(params): axum::extract::Query<HashMap<String, String>>,
        state: Arc<OrchestratorState>,
    ) -> impl IntoResponse {
        let query_str = params.get("q").cloned().unwrap_or_default();
        let category = params.get("category").cloned();
        let sort = params.get("sort").map(|s| match s.as_str() {
            "stars" => SortBy::Stars,
            "recent" => SortBy::Recent,
            "name" => SortBy::Name,
            _ => SortBy::Downloads,
        });

        // Set when the registry was tried and failed, so the local answer can say why.
        let mut registry_error: Option<String> = None;

        // Try registry search first
        let marketplace = state.marketplace.read().await;
        if let Some(client) = marketplace.as_ref() {
            let search_query = SearchQuery {
                query: if query_str.is_empty() {
                    None
                } else {
                    Some(query_str.clone())
                },
                category: category.clone(),
                keyword: None,
                author: None,
                sort_by: sort,
                page: params.get("page").and_then(|p| p.parse().ok()),
                per_page: params.get("per_page").and_then(|p| p.parse().ok()),
            };

            match client.search(search_query).await {
                Ok(results) => {
                    let agents: Vec<JsonValue> = results
                        .results
                        .iter()
                        .map(|a| {
                            json!({
                                "id": a.manifest.name,
                                "name": a.manifest.display_name,
                                "description": a.manifest.description,
                                "author": a.manifest.author.name,
                                "version": a.manifest.version,
                                "downloads": a.downloads,
                                "stars": a.stars,
                                "forks": a.forks,
                                "tags": a.manifest.keywords,
                                "updatedAt": a.updated_at * 1000,
                                "verified": a.verified,
                            })
                        })
                        .collect();

                    return Json(json!({
                        "success": true,
                        "query": query_str,
                        "category": category,
                        "agents": agents,
                        "total": results.total,
                        "page": results.page,
                        "per_page": results.per_page,
                    }));
                }
                Err(e) => {
                    // Fall through to the local-only response, which carries the error
                    registry_error = Some(e.to_string());
                }
            }
        }

        // Fallback: return installed agents that match the query
        let installed_agents = get_installed_as_json(&state).await;
        let filtered: Vec<JsonValue> = if query_str.is_empty() {
            installed_agents.clone()
        } else {
            let q = query_str.to_lowercase();
            installed_agents
                .iter()
                .filter(|a| {
                    a.get("name")
                        .and_then(|n| n.as_str())
                        .unwrap_or("")
                        .to_lowercase()
                        .contains(&q)
                        || a.get("description")
                            .and_then(|d| d.as_str())
                            .unwrap_or("")
                            .to_lowercase()
                            .contains(&q)
                })
                .cloned()
                .collect()
        };

        Json(json!({
            "success": true,
            "query": query_str,
            "category": category,
            "agents": filtered,
            "total": filtered.len(),
            "page": 1,
            "per_page": 20,
            "source": "local",
            "registry_error": registry_error,
        }))
    }

    async fn handle_marketplace_list(state: Arc<OrchestratorState>) -> impl IntoResponse {
        let installed_agents = get_installed_as_json(&state).await;

        Json(json!({
            "success": true,
            "agents": installed_agents,
            "total": installed_agents.len(),
        }))
    }

    /// A name or version the registry client refuses (`E_BAD_ARG`), answered
    /// as 400 before the client is consulted. `None` when both are safe.
    fn marketplace_name_refusal(
        name: &str,
        version: Option<&str>,
    ) -> Option<(StatusCode, Json<JsonValue>)> {
        let checked =
            crate::marketplace::validate_package_name(name, "agent name").and_then(|()| {
                match version {
                    Some(v) => crate::marketplace::validate_package_name(v, "agent version"),
                    None => Ok(()),
                }
            });
        let err = checked.err()?;
        Some((
            StatusCode::BAD_REQUEST,
            Json(json!({
                "success": false,
                "code": "E_BAD_ARG",
                "error": err.to_string(),
            })),
        ))
    }

    async fn handle_marketplace_install(
        Json(body): Json<JsonValue>,
        state: Arc<OrchestratorState>,
    ) -> impl IntoResponse {
        let name = match body.get("name").and_then(|n| n.as_str()) {
            Some(n) => n.to_string(),
            None => {
                return (
                    StatusCode::BAD_REQUEST,
                    Json(json!({
                        "success": false,
                        "error": "Missing required field: name"
                    })),
                )
            }
        };
        let version = body
            .get("version")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());
        if let Some(refusal) = marketplace_name_refusal(&name, version.as_deref()) {
            return refusal;
        }

        let mut marketplace = state.marketplace.write().await;
        let client = match marketplace.as_mut() {
            Some(c) => c,
            None => {
                return (
                    StatusCode::SERVICE_UNAVAILABLE,
                    Json(json!({
                        "success": false,
                        "error": "Marketplace client not available"
                    })),
                )
            }
        };

        match client.install(&name, version.as_deref()).await {
            Ok(installed) => {
                // Broadcast install event
                state
                    .broadcast_to_channel(
                        "agents",
                        json!({
                            "type": "agent_installed",
                            "agent": {
                                "name": installed.manifest.name,
                                "version": installed.manifest.version,
                                "description": installed.manifest.description,
                            }
                        }),
                    )
                    .await;

                (
                    StatusCode::OK,
                    Json(json!({
                        "success": true,
                        "message": format!("Installed {} v{}", installed.manifest.display_name, installed.manifest.version),
                        "agent": {
                            "name": installed.manifest.name,
                            "version": installed.manifest.version,
                            "path": installed.path.to_string_lossy(),
                        }
                    })),
                )
            }
            Err(e) => (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({
                    "success": false,
                    "error": format!("Failed to install {}: {}", name, e)
                })),
            ),
        }
    }

    async fn handle_marketplace_uninstall(
        Json(body): Json<JsonValue>,
        state: Arc<OrchestratorState>,
    ) -> impl IntoResponse {
        let name = match body.get("name").and_then(|n| n.as_str()) {
            Some(n) => n.to_string(),
            None => {
                return (
                    StatusCode::BAD_REQUEST,
                    Json(json!({
                        "success": false,
                        "error": "Missing required field: name"
                    })),
                )
            }
        };
        if let Some(refusal) = marketplace_name_refusal(&name, None) {
            return refusal;
        }

        let mut marketplace = state.marketplace.write().await;
        let client = match marketplace.as_mut() {
            Some(c) => c,
            None => {
                return (
                    StatusCode::SERVICE_UNAVAILABLE,
                    Json(json!({
                        "success": false,
                        "error": "Marketplace client not available"
                    })),
                )
            }
        };

        match client.uninstall(&name) {
            Ok(()) => {
                state
                    .broadcast_to_channel(
                        "agents",
                        json!({
                            "type": "agent_uninstalled",
                            "name": name,
                        }),
                    )
                    .await;

                (
                    StatusCode::OK,
                    Json(json!({
                        "success": true,
                        "message": format!("Uninstalled {}", name)
                    })),
                )
            }
            Err(e) => (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({
                    "success": false,
                    "error": format!("Failed to uninstall {}: {}", name, e)
                })),
            ),
        }
    }

    /// Helper: convert installed agents to JSON array
    async fn get_installed_as_json(state: &OrchestratorState) -> Vec<JsonValue> {
        let marketplace = state.marketplace.read().await;
        match marketplace.as_ref() {
            Some(client) => client
                .list_installed()
                .iter()
                .map(|a| {
                    json!({
                        "id": a.manifest.name,
                        "name": a.manifest.display_name,
                        "description": a.manifest.description,
                        "author": a.manifest.author.name,
                        "version": a.manifest.version,
                        "downloads": 0,
                        "stars": 0,
                        "forks": 0,
                        "tags": a.manifest.keywords,
                        "updatedAt": a.installed_at * 1000,
                        "verified": false,
                        "installed": true,
                    })
                })
                .collect(),
            None => vec![],
        }
    }

    #[cfg(test)]
    mod lifecycle_tests {
        use super::*;
        use std::time::Duration;

        /// With its one slot held past the wait, a second request is answered
        /// 503 with a code instead of queueing without end; once the slot is
        /// free, requests are admitted again.
        #[tokio::test]
        async fn a_request_that_cannot_get_a_slot_in_time_is_answered_busy() {
            crate::tls::install_provider();
            let permits = Arc::new(tokio::sync::Semaphore::new(1));
            let app = Router::new()
                .route(
                    "/slow",
                    get(|| async {
                        tokio::time::sleep(Duration::from_millis(1500)).await;
                        "slow"
                    }),
                )
                .route("/fast", get(|| async { "fast" }))
                .layer(axum::middleware::from_fn(move |req, next| {
                    admit(permits.clone(), 1, Duration::from_millis(200), req, next)
                }));
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = listener.local_addr().unwrap();
            let server = tokio::spawn(async move {
                let _ = axum::serve(listener, app).await;
            });
            let client = reqwest::Client::new();

            let slow = tokio::spawn({
                let client = client.clone();
                async move { client.get(format!("http://{addr}/slow")).send().await }
            });
            tokio::time::sleep(Duration::from_millis(300)).await;
            let busy = client
                .get(format!("http://{addr}/fast"))
                .send()
                .await
                .unwrap();
            assert_eq!(busy.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
            assert_eq!(busy.headers()["retry-after"], "5");
            let body: serde_json::Value = busy.json().await.unwrap();
            assert_eq!(body["code"], "E_BUDGET_EXCEEDED", "{body}");

            assert_eq!(slow.await.unwrap().unwrap().text().await.unwrap(), "slow");
            let fast = client
                .get(format!("http://{addr}/fast"))
                .send()
                .await
                .unwrap();
            assert_eq!(fast.status(), reqwest::StatusCode::OK);
            server.abort();
        }

        fn test_state() -> OrchestratorState {
            OrchestratorState {
                agents: RwLock::new(HashMap::new()),
                channels: RwLock::new(HashMap::new()),
                task_queue: RwLock::new(Vec::new()),
                workflows: RwLock::new(HashMap::new()),
                marketplace: RwLock::new(None),
            }
        }

        /// A task that never finishes on its own. The returned receiver
        /// resolves (with an error) once the task's future has been dropped,
        /// which is the point at which the task has really stopped.
        fn endless_task() -> (JoinHandle<()>, tokio::sync::oneshot::Receiver<()>) {
            let (dropped_tx, dropped_rx) = tokio::sync::oneshot::channel::<()>();
            let task = tokio::spawn(async move {
                let _sentinel = dropped_tx;
                std::future::pending::<()>().await
            });
            (task, dropped_rx)
        }

        async fn assert_stopped(dropped: tokio::sync::oneshot::Receiver<()>) {
            let outcome = tokio::time::timeout(Duration::from_secs(5), dropped)
                .await
                .expect("task future was never dropped");
            assert!(outcome.is_err(), "sentinel should drop without a value");
        }

        #[tokio::test]
        async fn unsubscribe_stops_the_forwarder() {
            let mut subs = SubscriptionTasks::default();
            let (task, dropped) = endless_task();
            subs.insert("news".to_string(), task);
            assert_eq!(subs.len(), 1);

            assert!(subs.remove("news").await);
            assert!(!subs.remove("news").await, "second remove finds nothing");
            assert_eq!(subs.len(), 0);
            assert_stopped(dropped).await;
        }

        #[tokio::test]
        async fn resubscribing_replaces_the_previous_forwarder() {
            let mut subs = SubscriptionTasks::default();
            let (first, first_dropped) = endless_task();
            let (second, _second_dropped) = endless_task();
            subs.insert("news".to_string(), first);
            subs.insert("news".to_string(), second);

            assert_eq!(subs.len(), 1);
            assert_stopped(first_dropped).await;
        }

        #[tokio::test]
        async fn disconnect_stops_every_forwarder() {
            let mut subs = SubscriptionTasks::default();
            let (a, a_dropped) = endless_task();
            let (b, b_dropped) = endless_task();
            subs.insert("a".to_string(), a);
            subs.insert("b".to_string(), b);

            subs.stop_all().await;

            assert_eq!(subs.len(), 0);
            assert_stopped(a_dropped).await;
            assert_stopped(b_dropped).await;
        }

        #[tokio::test]
        async fn dropping_the_set_aborts_what_is_left() {
            let mut subs = SubscriptionTasks::default();
            let (task, dropped) = endless_task();
            subs.insert("news".to_string(), task);
            drop(subs);
            assert_stopped(dropped).await;
        }

        #[tokio::test]
        async fn forwarder_skips_lagged_messages_and_ends_when_channel_closes() {
            // Capacity 2 and five sends: the receiver lags past 0..=2 and
            // should still deliver the two newest messages.
            let (tx, rx) = broadcast::channel(2);
            for i in 0..5 {
                let _ = tx.send(json!(i));
            }
            drop(tx);

            let (out_tx, mut out_rx) = mpsc::channel(16);
            let task = tokio::spawn(forward_channel(rx, out_tx, "news".to_string()));
            tokio::time::timeout(Duration::from_secs(5), task)
                .await
                .expect("forwarder did not end when the channel closed")
                .expect("forwarder panicked");

            let mut payloads = Vec::new();
            while let Some(msg) = out_rx.recv().await {
                match msg {
                    WsServerMessage::Channel { channel, payload } => {
                        assert_eq!(channel, "news");
                        payloads.push(payload);
                    }
                    _ => panic!("unexpected message kind"),
                }
            }
            assert_eq!(payloads, vec![json!(3), json!(4)]);
        }

        #[tokio::test]
        async fn channel_with_no_subscribers_is_pruned_on_broadcast() {
            let state = test_state();
            let idle = state.subscribe_to_channel("idle").await;
            let live = state.subscribe_to_channel("live").await;
            drop(idle);

            state.broadcast_to_channel("idle", json!(1)).await;

            let channels = state.channels.read().await;
            assert!(!channels.contains_key("idle"), "idle channel kept");
            assert!(channels.contains_key("live"), "live channel removed");
            drop(channels);
            drop(live);
        }

        async fn into_status_and_json<R: axum::response::IntoResponse>(
            response: R,
        ) -> (StatusCode, JsonValue) {
            let response = axum::response::IntoResponse::into_response(response);
            let status = response.status();
            let bytes = axum::body::to_bytes(response.into_body(), 1 << 20)
                .await
                .expect("response body");
            let body = serde_json::from_slice(&bytes).expect("JSON body");
            (status, body)
        }

        #[tokio::test]
        async fn create_workflow_refuses_malformed_steps_and_stores_nothing() {
            let state = Arc::new(test_state());
            let (status, body) = into_status_and_json(
                handle_create_workflow(
                    Json(json!({"name": "w", "steps": "not a list"})),
                    state.clone(),
                )
                .await,
            )
            .await;
            assert_eq!(status, StatusCode::BAD_REQUEST, "{body}");
            assert_eq!(body["success"], false, "{body}");
            assert!(state.workflows.read().await.is_empty());
        }

        #[tokio::test]
        async fn create_workflow_without_steps_is_still_created() {
            let state = Arc::new(test_state());
            let (status, body) = into_status_and_json(
                handle_create_workflow(Json(json!({"name": "w"})), state.clone()).await,
            )
            .await;
            assert_eq!(status, StatusCode::CREATED, "{body}");
            let id = body["workflow_id"].as_str().expect("workflow_id");
            let workflows = state.workflows.read().await;
            assert_eq!(workflows.len(), 1);
            assert!(workflows.contains_key(id));
        }

        #[tokio::test]
        async fn publish_says_the_agent_was_not_stored() {
            let state = Arc::new(test_state());
            let (status, body) = into_status_and_json(
                handle_publish_agent(Json(json!({"name": "a"})), state.clone()).await,
            )
            .await;
            assert_eq!(status, StatusCode::ACCEPTED, "{body}");
            assert_eq!(body["success"], true, "{body}");
            assert_eq!(body["stored"], false, "{body}");
        }

        #[tokio::test]
        async fn search_reports_a_failed_registry_in_registry_error() {
            // Loopback port 1 refuses at once, so no real registry is contacted.
            let state = Arc::new(OrchestratorState {
                marketplace: RwLock::new(Some(
                    RegistryClient::new(Some("http://127.0.0.1:1".to_string()))
                        .expect("registry client"),
                )),
                ..test_state()
            });
            let (status, body) = into_status_and_json(
                handle_marketplace_search(axum::extract::Query(HashMap::new()), state.clone())
                    .await,
            )
            .await;
            assert_eq!(status, StatusCode::OK, "{body}");
            assert_eq!(body["success"], true, "{body}");
            assert_eq!(body["source"], "local", "{body}");
            assert!(
                body["registry_error"]
                    .as_str()
                    .is_some_and(|e| !e.is_empty()),
                "{body}"
            );
        }

        #[tokio::test]
        async fn search_with_no_registry_has_no_registry_error() {
            let state = Arc::new(test_state());
            let (status, body) = into_status_and_json(
                handle_marketplace_search(axum::extract::Query(HashMap::new()), state.clone())
                    .await,
            )
            .await;
            assert_eq!(status, StatusCode::OK, "{body}");
            assert!(body["registry_error"].is_null(), "{body}");
        }

        /// A traversal name is refused as 400 (`E_BAD_ARG`) on install and
        /// uninstall, with no registry contacted and nothing written. The
        /// client points at a dead loopback port, so a request would fail as
        /// a network error instead.
        #[tokio::test]
        async fn marketplace_refuses_a_traversal_name_as_bad_arg() {
            let cache = std::env::temp_dir().join(format!(
                "ish-api-mp-{}-{}",
                std::process::id(),
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_nanos()
            ));
            let client = crate::marketplace::RegistryClient::with_cache_dir(
                Some("http://127.0.0.1:1".to_string()),
                cache.clone(),
            )
            .expect("registry client over a temp cache");
            let state = Arc::new(OrchestratorState {
                marketplace: RwLock::new(Some(client)),
                ..test_state()
            });
            for body in [
                json!({"name": "../../escape"}),
                json!({"name": "a/b"}),
                json!({"name": "ok", "version": "..\\x"}),
            ] {
                let (status, out) = into_status_and_json(
                    handle_marketplace_install(Json(body.clone()), state.clone()).await,
                )
                .await;
                assert_eq!(status, StatusCode::BAD_REQUEST, "{body} -> {out}");
                assert_eq!(out["code"], "E_BAD_ARG", "{out}");
            }
            let (status, out) = into_status_and_json(
                handle_marketplace_uninstall(Json(json!({"name": "../x"})), state.clone()).await,
            )
            .await;
            assert_eq!(status, StatusCode::BAD_REQUEST, "{out}");
            assert_eq!(out["code"], "E_BAD_ARG", "{out}");

            // With no client configured the refusal still comes first.
            let bare = Arc::new(test_state());
            let (status, out) = into_status_and_json(
                handle_marketplace_install(Json(json!({"name": ".."})), bare).await,
            )
            .await;
            assert_eq!(status, StatusCode::BAD_REQUEST, "{out}");

            // Nothing was written into the cache.
            assert_eq!(std::fs::read_dir(&cache).unwrap().count(), 0);
            let _ = std::fs::remove_dir_all(&cache);
        }

        #[tokio::test]
        async fn evaluate_off_thread_returns_the_evaluation_result() {
            let response = evaluate_off_thread(
                AgentRequest::Eval {
                    code: "1 + 1".to_string(),
                },
                None,
            )
            .await
            .expect("evaluation panicked");
            assert!(response.success);
        }

        /// The stream bound refuses at once once every slot is held, and a
        /// released slot is available again. The handlers call this same
        /// helper before they evaluate, so a mutation that drops the check
        /// there would leave the limit unenforced.
        #[test]
        fn stream_slots_refuse_when_full_and_recover_on_release() {
            let slots = Arc::new(tokio::sync::Semaphore::new(2));
            let a = try_take_stream_slot(&slots).expect("first slot");
            let b = try_take_stream_slot(&slots).expect("second slot");
            assert!(
                try_take_stream_slot(&slots).is_none(),
                "a third streamed evaluation must be refused, not queued"
            );
            drop(a);
            assert!(
                try_take_stream_slot(&slots).is_some(),
                "a released slot must be reusable"
            );
            drop(b);
        }

        #[test]
        fn a_refused_stream_is_a_coded_busy_response() {
            let r = ws_busy_response();
            assert!(!r.success);
            assert_eq!(r.code.as_deref(), Some("E_BUDGET_EXCEEDED"));
        }
    }
}

// ============================================================================
// CLI Integration
// ============================================================================

/// Process a JSON request from stdin (for CLI usage)
pub fn process_json_request(json_str: &str) -> Result<String> {
    // The same parse as `parse_request`, so a repeated key is refused here too.
    let request = parse_request(json_str).map_err(|r| {
        anyhow!(
            "{}",
            r.error
                .unwrap_or_else(|| "Invalid JSON request".to_string())
        )
    })?;

    let response = process_request(&request);
    Ok(serde_json::to_string_pretty(&response)?)
}

/// The hint for input that is not an agent request.
const REQUEST_HINT: &str = "a request is {\"action\":\"call\",\"builtin\":NAME,\"args\":[...]}; actions: call, pipeline, eval, describe, list_builtins, schema, type_info";

/// The first key an object repeats anywhere in a JSON text, as a path such as
/// `args.path`. `serde_json::Value` keeps only the last of a repeated key, so a
/// `Value` cannot show that a request repeated one; this walks the text itself.
struct FirstRepeatedKey(Option<String>);

/// Walks one JSON value, noting the first repeated key under `path`.
struct RepeatSeed<'p> {
    path: &'p str,
}

impl<'de> serde::de::DeserializeSeed<'de> for RepeatSeed<'_> {
    type Value = Option<String>;

    fn deserialize<D: serde::Deserializer<'de>>(self, d: D) -> Result<Option<String>, D::Error> {
        d.deserialize_any(RepeatVisitor { path: self.path })
    }
}

struct RepeatVisitor<'p> {
    path: &'p str,
}

impl<'de> serde::de::Visitor<'de> for RepeatVisitor<'_> {
    type Value = Option<String>;

    fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        f.write_str("JSON")
    }

    fn visit_bool<E>(self, _: bool) -> Result<Option<String>, E> {
        Ok(None)
    }
    fn visit_i64<E>(self, _: i64) -> Result<Option<String>, E> {
        Ok(None)
    }
    fn visit_u64<E>(self, _: u64) -> Result<Option<String>, E> {
        Ok(None)
    }
    fn visit_f64<E>(self, _: f64) -> Result<Option<String>, E> {
        Ok(None)
    }
    fn visit_str<E>(self, _: &str) -> Result<Option<String>, E> {
        Ok(None)
    }
    fn visit_unit<E>(self) -> Result<Option<String>, E> {
        Ok(None)
    }

    fn visit_seq<A: serde::de::SeqAccess<'de>>(
        self,
        mut seq: A,
    ) -> Result<Option<String>, A::Error> {
        let mut found = None;
        let mut i = 0usize;
        loop {
            let path = format!("{}[{i}]", self.path);
            match seq.next_element_seed(RepeatSeed { path: &path })? {
                Some(inner) => {
                    if found.is_none() {
                        found = inner;
                    }
                }
                None => break,
            }
            i += 1;
        }
        Ok(found)
    }

    fn visit_map<A: serde::de::MapAccess<'de>>(
        self,
        mut map: A,
    ) -> Result<Option<String>, A::Error> {
        let mut seen = std::collections::HashSet::new();
        let mut found = None;
        while let Some(key) = map.next_key::<String>()? {
            let path = if self.path.is_empty() {
                key.clone()
            } else {
                format!("{}.{key}", self.path)
            };
            let inner = map.next_value_seed(RepeatSeed { path: &path })?;
            if found.is_none() {
                if !seen.insert(key) {
                    found = Some(path);
                } else {
                    found = inner;
                }
            }
        }
        Ok(found)
    }
}

/// The path of the first key an object in `text` repeats, if any. `None` also
/// when `text` is not JSON; the caller's own parse reports that.
pub(crate) fn first_repeated_key(text: &str) -> Option<String> {
    serde_json::from_str::<FirstRepeatedKey>(text)
        .ok()
        .and_then(|k| k.0)
}

impl<'de> serde::Deserialize<'de> for FirstRepeatedKey {
    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        serde::de::DeserializeSeed::deserialize(RepeatSeed { path: "" }, d).map(FirstRepeatedKey)
    }
}

/// Parse one request from its JSON text. A leading byte-order mark is ignored.
/// Text that is not JSON is `E_PARSE`, and JSON that is not a request is
/// `E_BAD_ARG`. Either is returned as the failure response itself, so a caller
/// always has exactly one response to print.
pub fn parse_request(text: &str) -> Result<AgentRequest, AgentResponse> {
    let text = text.strip_prefix('\u{feff}').unwrap_or(text);
    let value: JsonValue = serde_json::from_str(text).map_err(|e| {
        // Text after the first request is what a second request on the same
        // input looks like, so the hint says how requests are given.
        let hint = if e.to_string().starts_with("trailing characters") {
            several_requests_hint()
        } else if names_a_file(text) {
            format!(
                "the argument is the name of a file, but execute takes the JSON text itself: pass its contents (`ish agent execute \"$(cat FILE)\"`) or pipe the file to stdin. {REQUEST_HINT}"
            )
        } else {
            REQUEST_HINT.to_string()
        };
        bad_request("E_PARSE", format!("Invalid JSON request: {e}"), hint)
    })?;
    // A key the request repeats is refused, not resolved to its last value: a
    // reader of the text and the parser could disagree about which one counts.
    if value.is_object() {
        if let Ok(FirstRepeatedKey(Some(repeated))) = serde_json::from_str::<FirstRepeatedKey>(text)
        {
            return Err(bad_request(
                "E_PARSE",
                format!("Invalid JSON request: duplicate key \"{repeated}\""),
                REQUEST_HINT.to_string(),
            ));
        }
    }
    // A pipeline that does not parse is most often written with `stages` for
    // `steps`, so its hint shows the shape.
    let hint = if value.get("action").and_then(JsonValue::as_str) == Some("pipeline") {
        PIPELINE_HINT.to_string()
    } else {
        REQUEST_HINT.to_string()
    };
    serde_json::from_value(value)
        .map_err(|e| bad_request("E_BAD_ARG", format!("Invalid request: {e}"), hint))
}

/// Whether `text` is one short line naming an existing file inside the working
/// folder. Only a relative path without `..` is looked at: an absolute or UNC
/// name is never probed, so this reads nothing outside the folder.
fn names_a_file(text: &str) -> bool {
    let t = text.trim();
    if t.is_empty() || t.len() > 260 || t.contains(['\n', '{', '[', '"']) {
        return false;
    }
    let p = std::path::Path::new(t);
    if p.is_absolute() || t.starts_with(['\\', '/']) {
        return false;
    }
    if p.components()
        .any(|c| !matches!(c, std::path::Component::Normal(_)))
    {
        return false;
    }
    p.is_file()
}

/// The hint for a pipeline request that does not parse: its shape, whose list
/// is named `steps`.
const PIPELINE_HINT: &str = "a pipeline is {\"action\":\"pipeline\",\"steps\":[{\"builtin\":NAME,\"args\":[...]}]}, with an optional \"input\"; the list is named steps, not stages";

/// The hint for input that holds more than one request.
fn several_requests_hint() -> String {
    format!(
        "execute takes one request; `ish agent interactive` takes one request per line. {REQUEST_HINT}"
    )
}

/// The failure response for input that is not a request.
fn bad_request(code: &str, message: String, hint: String) -> AgentResponse {
    AgentResponse {
        success: false,
        result: None,
        error: Some(message),
        code: Some(code.to_string()),
        hint: Some(hint),
        result_type: None,
        metadata: None,
    }
}

/// The one response to a request's JSON text: the result of the request, or the
/// failure for input that is not one. For a caller that prints a response per
/// request, as `ish agent execute` and `ish agent interactive` do.
pub fn process_json_response(text: &str) -> AgentResponse {
    match parse_request(text) {
        Ok(request) => process_request(&request),
        Err(response) => response,
    }
}

/// The names `ish agent schema <format>` and `GET /api/v1/schema/:format`
/// accept, matched case-insensitively. None for a name neither knows, so
/// that a typo is reported rather than answered with some other format.
pub fn parse_schema_format(format: &str) -> Option<SchemaFormat> {
    match format.to_lowercase().as_str() {
        // OpenAI family
        "openai" | "gpt" | "chatgpt" | "open_a_i" => Some(SchemaFormat::OpenAI),
        "azure" | "azure_openai" | "azureopenai" | "azure_open_a_i" => {
            Some(SchemaFormat::AzureOpenAI)
        }

        // Anthropic
        "claude" | "anthropic" => Some(SchemaFormat::Claude),

        // Google
        "gemini" | "google" => Some(SchemaFormat::Gemini),

        // Meta
        "llama" | "meta" | "llama3" => Some(SchemaFormat::Llama),

        // Mistral
        "mistral" | "codestral" | "pixtral" => Some(SchemaFormat::Mistral),

        // Cohere
        "cohere" | "command" | "command-r" => Some(SchemaFormat::Cohere),

        // xAI
        "grok" | "xai" => Some(SchemaFormat::Grok),

        // DeepSeek
        "deepseek" | "deepseek-r1" | "deep_seek" => Some(SchemaFormat::DeepSeek),

        // AWS
        "bedrock" | "aws" | "amazon" => Some(SchemaFormat::Bedrock),

        // Alibaba
        "qwen" | "alibaba" | "dashscope" => Some(SchemaFormat::Qwen),

        // Local/Self-hosted
        "ollama" => Some(SchemaFormat::Ollama),
        // LM Studio serves the OpenAI chat-completions API.
        "lmstudio" | "lm-studio" | "lm_studio" => Some(SchemaFormat::OpenAI),
        "vllm" | "v_llm" => Some(SchemaFormat::VLLM),

        // Platforms
        "huggingface" | "hugging_face" | "hf" | "tgi" => Some(SchemaFormat::HuggingFace),
        "openrouter" | "open_router" => Some(SchemaFormat::OpenRouter),

        // Chinese AI providers
        "kimi" | "moonshot" => Some(SchemaFormat::Kimi),
        "yi" | "01ai" | "lingyiwanwu" => Some(SchemaFormat::Yi),
        "glm" | "g_l_m" | "chatglm" | "zhipu" => Some(SchemaFormat::GLM),

        // Additional SOTA providers
        "reka" => Some(SchemaFormat::Reka),
        "ai21" | "a_i21" | "jamba" | "jurassic" => Some(SchemaFormat::AI21),
        "perplexity" | "sonar" => Some(SchemaFormat::Perplexity),
        "together" | "together-ai" => Some(SchemaFormat::Together),
        "groq" => Some(SchemaFormat::Groq),
        "fireworks" | "fireworks-ai" => Some(SchemaFormat::Fireworks),

        // Standard formats
        "json" | "jsonschema" | "json_schema" | "full" => Some(SchemaFormat::JsonSchema),
        "compact" | "ontology" => Some(SchemaFormat::Ontology),
        "jsonld" | "json-ld" | "json_ld" | "linked_data" => Some(SchemaFormat::JsonLD),
        "owl" | "rdf" | "turtle" | "owl_turtle" => Some(SchemaFormat::OwlTurtle),
        "shacl" | "shapes" | "s_h_a_c_l" => Some(SchemaFormat::SHACL),

        _ => None,
    }
}

/// The format names the error for an unknown one lists.
pub const SCHEMA_FORMATS_HINT: &str = "openai, claude, gemini, llama, mistral, cohere, grok, deepseek, bedrock, azure, azure_openai, qwen, ollama, lmstudio, vllm, huggingface, openrouter, kimi, yi, glm, reka, ai21, perplexity, together, groq, fireworks, json, compact, ontology, jsonld, owl, shacl";

/// Generate schema in the specified format
pub fn generate_schema(format: &str) -> Result<String> {
    generate_schema_filtered(format, None, None)
}

/// Generate schema in the specified format, narrowed as `GET /api/v1/schema`
/// narrows it with `module` and `q`.
pub fn generate_schema_filtered(
    format: &str,
    module: Option<&str>,
    q: Option<&str>,
) -> Result<String> {
    let schema_format = parse_schema_format(format).ok_or_else(|| {
        anyhow!(
            "Unknown schema format: '{}'. Supported: {}",
            format,
            SCHEMA_FORMATS_HINT
        )
    })?;

    let request = AgentRequest::Schema {
        format: schema_format,
        module: module.map(str::to_string),
        q: q.map(str::to_string),
    };
    let response = process_request(&request);

    if response.success {
        Ok(serde_json::to_string_pretty(&response.result)?)
    } else {
        Err(anyhow!(
            "Failed to generate schema: {}",
            response.error.unwrap_or_default()
        ))
    }
}