ocserve_http/

lib.rs

//! ocserve-http: axum router, auth, byte-golden SSE, upstream error envelope.
#![deny(unsafe_code)]

use axum::extract::State;
use axum::http::{HeaderValue, StatusCode};
use axum::response::sse::{Event, Sse};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use axum::{Json, Router};
use futures_util::StreamExt;
use futures_util::stream;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

/// Freeze (PLAN §3): version reported by `/global/health`.
pub const FREEZE_VERSION: &str = "1.18.31";

/// Pre-assembled payloads for config-derived routes (built by ocserve's
/// `Runtime::load()`; kept crate-local so http doesn't depend on cli).
/// JSON deep merge: objects merge recursively, everything else (arrays,
/// scalars) replaces. Used by overlay layering + PATCH /config (W4).
pub fn deep_merge(base: &mut serde_json::Value, patch: serde_json::Value) {
    match (base, patch) {
        (serde_json::Value::Object(b), serde_json::Value::Object(p)) => {
            for (k, v) in p {
                match b.get_mut(&k) {
                    Some(bv) => deep_merge(bv, v),
                    None => {
                        b.insert(k, v);
                    }
                }
            }
        }
        (b, p) => *b = p,
    }
}

/// W4 config-write target: default = the SHARED opencode.json (drop-in
/// contract — both servers read it); `OCSERVE_CONFIG_WRITE=overlay` switches
/// to a ocserve-owned file that layers over the base at load (never touches
/// the other server's config).
pub fn config_write_path() -> std::path::PathBuf {
    let home = std::env::var("HOME").unwrap_or_else(|_| "/".into());
    if std::env::var("OCSERVE_CONFIG_WRITE")
        .map(|v| v == "overlay")
        .unwrap_or(false)
    {
        std::path::PathBuf::from(home).join(".config/ocserve/config.json")
    } else {
        std::path::PathBuf::from(home).join(".config/opencode/opencode.json")
    }
}

/// Boot-injected config reloader (W4): Runtime::load lives in ocserve —
/// the closure avoids a crate cycle; None in tests. Returns derived payloads
/// AND a rebuilt LLM registry (H2 — auth/provider/default-model edits apply
/// without restart). Callers go through `watch::reconcile`, never the raw fn.
pub type ConfigReloader =
    std::sync::Arc<dyn Fn() -> anyhow::Result<(Payloads, LlmRegistry)> + Send + Sync>;

#[derive(Default)]
pub struct Payloads {
    pub config: serde_json::Value,
    pub agent: Vec<serde_json::Value>,
    pub api_agent: Vec<serde_json::Value>,
    pub command: Vec<serde_json::Value>,
    pub config_providers: serde_json::Value,
    pub provider: serde_json::Value,
    pub console: serde_json::Value,
    pub capabilities: serde_json::Value,
    /// shared-config `compaction` section (M6 trigger/knobs).
    pub compaction: serde_json::Value,
    /// Parsed models catalog (`models.json`, ~5.4 MB), swapped with the other
    /// payloads on reload. Shared so v2 metadata routes never re-read +
    /// re-parse the file per request — the pre-fix `read_catalog_value` did
    /// exactly that (100 ms CPU + ~25 MB retention per call; conc-6 → 8 s).
    pub catalog: std::sync::Arc<serde_json::Value>,
}

/// PERF-10X F1: run a blocking store call on the runtime's blocking pool.
/// Store fns are sync by design (STORAGE: never hold a connection across
/// `.await`) — calling them inline parks a tokio worker for the query's
/// duration. Baseline measured exactly that convoy: one 6s search held a
/// worker and every other route queued behind it (config p95 998ms for a
/// payload read; starvation signature: tail explodes while CPU ≪100%).
async fn run_blocking<T, F>(db: &std::path::Path, op: F) -> anyhow::Result<T>
where
    F: FnOnce(&std::path::Path) -> anyhow::Result<T> + Send + 'static,
    T: Send + 'static,
{
    let db = db.to_path_buf();
    tokio::task::spawn_blocking(move || op(&db))
        .await
        .map_err(|e| anyhow::anyhow!("blocking task join: {e}"))?
}

pub mod compress;
pub mod doc;
pub mod embedded_ui;
pub mod pty;
pub mod tui;
pub mod ui;
pub mod v2;
pub mod vcs;

/// Upstream error envelope: {"name":"NotFoundError","data":{"message":"..."}} (captured live).
pub struct ApiError {
    pub status: StatusCode,
    pub name: &'static str,
    pub message: String,
}

impl ApiError {
    pub fn not_found(msg: impl Into<String>) -> Self {
        Self {
            status: StatusCode::NOT_FOUND,
            name: "NotFoundError",
            message: msg.into(),
        }
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let body = json!({"name": self.name, "data": {"message": self.message}});
        (self.status, Json(body)).into_response()
    }
}

/// Handler error: normal envelopes + Effect HttpApi tagged decode errors
/// (freeze parity: `?before=` → body exactly {"_tag":"BadRequest"} §1090).
pub mod sync;
pub mod watch;

pub enum HttpError {
    Api(ApiError),
    Tagged {
        status: StatusCode,
        tag: &'static str,
    },
    /// Effect query-decode failure envelope (freeze probes: `?limit=abc` →
    /// {"name":"BadRequest","data":{"message":...,"kind":"Query"}}).
    Query {
        message: String,
    },
    /// Effect payload-decode failure envelope (freeze probes: missing body key
    /// → {"name":"BadRequest","data":{"message":"Missing key\n  at [...]",
    /// "kind":"Payload"}}).
    Payload {
        message: String,
    },
    /// Effect path-param decode failure envelope (freeze: bad `que_`/`per_`
    /// prefix → {"name":"BadRequest","data":{"message":"Expected a string
    /// starting with \"que\", got \"x\"\n  at [\"requestID\"]","kind":"Params"}}).
    Params {
        message: String,
    },
    /// Effect TaggedErrorClass envelope with schema fields at top level
    /// (errors.ts:143 McpServerNotFoundError → {"_tag":..., name, message}).
    TaggedData {
        status: StatusCode,
        tag: &'static str,
        fields: serde_json::Value,
    },
}

impl From<ApiError> for HttpError {
    fn from(e: ApiError) -> Self {
        HttpError::Api(e)
    }
}

impl IntoResponse for HttpError {
    fn into_response(self) -> Response {
        match self {
            HttpError::Api(e) => e.into_response(),
            HttpError::Tagged { status, tag } => {
                let body = format!("{{\"_tag\":\"{tag}\"}}");
                (
                    status,
                    [(axum::http::header::CONTENT_TYPE, "application/json")],
                    body,
                )
                    .into_response()
            }
            HttpError::TaggedData {
                status,
                tag,
                fields,
            } => {
                let mut obj = serde_json::Map::new();
                obj.insert("_tag".into(), serde_json::Value::String(tag.into()));
                if let serde_json::Value::Object(f) = fields {
                    obj.extend(f);
                }
                (
                    status,
                    [(axum::http::header::CONTENT_TYPE, "application/json")],
                    serde_json::Value::Object(obj).to_string(),
                )
                    .into_response()
            }
            HttpError::Query { message } => {
                let body = serde_json::json!({
                    "name": "BadRequest",
                    "data": {"message": message, "kind": "Query"}
                })
                .to_string();
                (
                    StatusCode::BAD_REQUEST,
                    [(axum::http::header::CONTENT_TYPE, "application/json")],
                    body,
                )
                    .into_response()
            }
            HttpError::Payload { message } => {
                let body = serde_json::json!({
                    "name": "BadRequest",
                    "data": {"message": message, "kind": "Payload"}
                })
                .to_string();
                (
                    StatusCode::BAD_REQUEST,
                    [(axum::http::header::CONTENT_TYPE, "application/json")],
                    body,
                )
                    .into_response()
            }
            HttpError::Params { message } => {
                let body = serde_json::json!({
                    "name": "BadRequest",
                    "data": {"message": message, "kind": "Params"}
                })
                .to_string();
                (
                    StatusCode::BAD_REQUEST,
                    [(axum::http::header::CONTENT_TYPE, "application/json")],
                    body,
                )
                    .into_response()
            }
        }
    }
}

/// session → (generation, background prompt task) for POST /abort.
pub type PromptTask = (u64, tokio::task::JoinHandle<()>);

/// App state shared by handlers.
pub struct AppState {
    /// Config-derived route payloads (W4: swappable after PATCH /config —
    /// upstream marks the instance for disposal and serves fresh config).
    pub payloads: parking_lot::RwLock<Payloads>,
    /// PERF-10X F5: pre-serialized bytes for the hot Json-clone handlers
    /// (config/agent/command/config_providers/provider/console/capabilities).
    /// Built at construction + every reload, so a request is a Bytes clone
    /// (refcount) + zero-copy Body — no per-request Value clone/serialize.
    /// Kill-switch: OCSERVE_WIRE_CACHE=0 (wire_off) falls back to the Value
    /// path; both paths serve serde-identical bytes (unit-tested).
    pub wire: parking_lot::RwLock<HashMap<&'static str, WireEntry>>,
    pub wire_off: std::sync::atomic::AtomicBool,
    /// Boot-injected config reloader (Runtime lives in ocserve; the
    /// closure avoids a crate cycle). None in tests → PATCH still writes
    /// the file, swap skipped (logged).
    pub reloader: parking_lot::RwLock<Option<ConfigReloader>>,
    /// Runtime paths for GET /path (env-derived at boot; never stored in repo).
    pub paths: serde_json::Value,
    /// Auth mode: None = off (freeze default), Some(("user","pass")) = basic.
    pub auth: Option<(String, String)>,
    /// Request counter for /metrics (bounded, monotonic).
    pub requests: std::sync::atomic::AtomicU64,
    // ---- M2 prompt runtime (wired by CLI) ----
    /// Bounded broadcast of encoded global-event frames.
    pub bus: ocserve_core::EventBus,
    pub db: std::path::PathBuf,
    pub blobs: std::sync::Arc<ocserve_store::BlobStore>,
    pub writer: std::sync::Arc<ocserve_store::Writer>,
    /// Provider registry + default model + agent systems (from Runtime).
    /// RwLock since H2: rebuilds on config hot-reload (watch::reconcile);
    /// in-flight prompts keep the values they resolved at build time.
    pub llm: parking_lot::RwLock<LlmRegistry>,
    /// Config hot-reload watch state (H1): polled paths + observed tuples.
    pub watch: parking_lot::RwLock<watch::WatchState>,
    /// Serializes reconcile across PATCH/auth/watcher callers (tokio Mutex —
    /// held across reload + MCP awaits; never a parking_lot guard over await).
    pub reconcile_lock: tokio::sync::Mutex<()>,
    /// K-AUTONOMY knobs — read from env ONCE at construction (changing
    /// OCSERVE_PROMPT_MAX_ROUNDS / OCSERVE_PROMPT_MAX_COST_USD = restart).
    pub max_rounds: usize,
    pub cost_ceiling: f64,
    /// Permission rendezvous shared by runner + reply routes.
    pub gate: std::sync::Arc<ocserve_core::PermissionGate>,
    /// MCP hub probed at serve boot (OnceLock: tests run without probing).
    pub mcp: std::sync::OnceLock<std::sync::Arc<ocserve_mcp::McpHub>>,
    /// Plugin sidecar (M4b): async mutex — trigger() takes &mut across await.
    pub plugins: std::sync::OnceLock<std::sync::Arc<tokio::sync::Mutex<ocserve_plugin::Sidecar>>>,
    /// Per-session prompt serialization (queue semantics). Bounded: entries
    /// exist only for in-flight prompts + transient races, removed post-run
    /// when uncontended (AGENTS §2.3 — active-set bound, no eviction).
    pub prompt_locks: std::sync::Arc<
        parking_lot::Mutex<
            std::collections::HashMap<String, std::sync::Arc<tokio::sync::Mutex<()>>>,
        >,
    >,
    /// Background prompt tasks per session: (generation, JoinHandle) —
    /// POST /abort cancels the current generation (v1 session.abort).
    /// Bound: one entry per in-flight async prompt, removed on completion.
    pub prompt_tasks:
        std::sync::Arc<parking_lot::Mutex<std::collections::HashMap<String, PromptTask>>>,
    /// Monotonic generation source for prompt_tasks.
    pub prompt_gen: std::sync::Arc<std::sync::atomic::AtomicU64>,
    /// Question rendezvous (v1 Question service): pending asks + reply/reject.
    pub question_gate: std::sync::Arc<ocserve_core::question::QuestionGate>,
    /// /tui/control rendezvous queues (v1 shared/tui-control.ts)
    pub tui: crate::tui::TuiControl,
    /// Web-UI reverse proxy (Phase 7; upstream serveUIEffect parity)
    pub ui: crate::ui::UiProxy,
    /// PTY sessions (/pty/* — web-UI terminal)
    pub pty: std::sync::Arc<ocserve_pty::PtyManager>,
    /// User CORS allowlist (config `server.cors` + `--cors`), read once at boot
    pub cors_extra: Vec<String>,
}

/// LLM endpoint resolution for the prompt runner (assembled by Runtime).
#[derive(Default)]
pub struct LlmRegistry {
    /// providerID → (base_url, api_key)
    pub endpoints: std::collections::HashMap<String, (String, String)>,
    /// (providerID, modelID) → (input, output, cache_read) USD/MTok
    pub pricing: std::collections::HashMap<(String, String), (f64, f64, f64)>,
    /// (providerID, modelID) → `limit` block (context/input) — M6 trigger
    pub limits: std::collections::HashMap<(String, String), serde_json::Value>,
    /// default (providerID, modelID) from opencode state
    pub default_model: (String, String),
    /// agent name → system prompt (only agents with explicit prompts)
    pub systems: std::collections::HashMap<String, String>,
    /// config default_agent (upstream session rows record it at prompt time)
    pub default_agent: String,
}

/// Real runtime pieces passed at construction (no interior mutability, no unsafe).
pub struct Wires {
    pub db: std::path::PathBuf,
    pub blobs: std::sync::Arc<ocserve_store::BlobStore>,
    pub writer: std::sync::Arc<ocserve_store::Writer>,
    pub llm: LlmRegistry,
}

/// One wire-cache entry: identity bytes, precomputed compressed variants,
/// and a strong ETag over the identity body (WEBUI-PLAN W1/W2).
///
/// Compressed variants are built once per write-epoch (reload), never on the
/// request path — the provider payload is 6.2 MB and brotli-q5 costs ~0.18 s.
#[derive(Clone)]
pub struct WireEntry {
    pub identity: bytes::Bytes,
    pub br: bytes::Bytes,
    pub gzip: bytes::Bytes,
    pub etag: String,
}

/// Serialize the hot payload fields once (reload-time). off => empty map
/// (handlers fall back to the Value path). `location` is the v2 envelope's
/// location object (built from `paths`, which is static per process).
pub(crate) fn rebuild_wire(
    p: &Payloads,
    off: bool,
    location: &Value,
) -> HashMap<&'static str, WireEntry> {
    let mut m = HashMap::new();
    if off {
        return m;
    }
    // serde_json::to_vec is exactly what axum's Json uses — byte-identical.
    fn one<T: serde::Serialize>(m: &mut HashMap<&'static str, WireEntry>, k: &'static str, v: &T) {
        if let Ok(b) = serde_json::to_vec(v) {
            let identity = bytes::Bytes::from(b);
            let etag = crate::compress::etag_for(&identity);
            let br = bytes::Bytes::from(crate::compress::compress(
                &identity,
                crate::compress::Encoding::Brotli,
            ));
            let gzip = bytes::Bytes::from(crate::compress::compress(
                &identity,
                crate::compress::Encoding::Gzip,
            ));
            m.insert(
                k,
                WireEntry {
                    identity,
                    br,
                    gzip,
                    etag,
                },
            );
        }
    }
    // /global/config == /config minus instance-scoped keys (freeze probe).
    {
        let mut gc = p.config.clone();
        if let Some(obj) = gc.as_object_mut() {
            for k in ["agent", "command", "mode", "username"] {
                obj.remove(k);
            }
        }
        one(&mut m, "global_config", &gc);
    }
    one(&mut m, "config", &p.config);
    one(&mut m, "agent", &p.agent);
    one(&mut m, "command", &p.command);
    one(&mut m, "config_providers", &p.config_providers);
    one(&mut m, "provider", &p.provider);
    one(&mut m, "console", &p.console);
    one(&mut m, "capabilities", &p.capabilities);
    // v2 metadata group: projection is pure over Payloads (the pre-fix
    // handlers re-read + re-parsed the 5.4 MB catalog and deep-cloned the
    // provider tree on EVERY request).
    one(&mut m, "api_model", &crate::v2::models_body(p, location));
    one(
        &mut m,
        "api_provider",
        &crate::v2::providers_body(p, location),
    );
    one(
        &mut m,
        "api_integration",
        &crate::v2::integrations_body(p, location),
    );
    // v2 agent/command lists (perf dive 2026-10-11): per-request projection +
    // serde on every UI refresh; pure over Payloads, so cache like the rest.
    one(&mut m, "api_agent", &api_agent_body(p, location));
    one(&mut m, "api_command", &api_command_body(p, location));
    m
}

/// Serve cached bytes as a JSON response with negotiation + validators; None
/// = fall back to the Value clone path (wire cache off / unknown key).
/// Identity clients (no Accept-Encoding, no If-None-Match) receive the exact
/// identity bytes and content-type as before (WEBUI-PLAN target 3).
fn wire_json_h(
    st: &AppState,
    key: &'static str,
    accept_encoding: Option<&str>,
    if_none_match: Option<&str>,
) -> Option<axum::response::Response> {
    if st.wire_off.load(std::sync::atomic::Ordering::Relaxed) {
        return None;
    }
    let e = st.wire.read().get(key).cloned()?;
    let encoding = crate::compress::negotiate(accept_encoding);
    // 304 short-circuit (RFC 9111): before any body work.
    if crate::compress::if_none_match_matches(if_none_match, &e.etag) {
        return Some(
            axum::response::Response::builder()
                .status(axum::http::StatusCode::NOT_MODIFIED)
                .header(axum::http::header::ETAG, &e.etag)
                .header(axum::http::header::VARY, "Accept-Encoding")
                .body(axum::body::Body::empty())
                .expect("static 304"),
        );
    }
    let mut b = axum::response::Response::builder()
        .header(axum::http::header::CONTENT_TYPE, "application/json")
        .header(axum::http::header::ETAG, &e.etag)
        .header(axum::http::header::VARY, "Accept-Encoding");
    let body = match encoding {
        Some(crate::compress::Encoding::Brotli) => {
            b = b.header(axum::http::header::CONTENT_ENCODING, "br");
            e.br.clone()
        }
        Some(crate::compress::Encoding::Gzip) => {
            b = b.header(axum::http::header::CONTENT_ENCODING, "gzip");
            e.gzip.clone()
        }
        None => e.identity.clone(),
    };
    Some(
        b.body(axum::body::Body::from(body))
            .expect("static response"),
    )
}

/// Back-compat shim (tests + callers with no header context): identity only.
#[cfg(test)]
fn wire_json(st: &AppState, key: &'static str) -> Option<axum::response::Response> {
    wire_json_h(st, key, None, None)
}

impl AppState {
    pub fn new() -> Arc<Self> {
        Self::with_auth(None)
    }

    pub fn with_auth(auth: Option<(String, String)>) -> Arc<Self> {
        Self::with_payloads(auth, Payloads::default())
    }

    /// Build from pre-assembled route payloads with a temp-backed store
    /// (tests / simple construction — real pieces, no placeholders).
    pub fn with_payloads(auth: Option<(String, String)>, p: Payloads) -> Arc<Self> {
        // pid + nanos + counter: parallel tests calling this on the same clock
        // tick must not share a temp data dir (writer/db contention → flaky).
        static NEW_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let dir = std::env::temp_dir().join(format!(
            "ocserve-http-{}-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0),
            NEW_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed),
        ));
        std::fs::create_dir_all(&dir).expect("temp data dir");
        let db = ocserve_store::writer::db_path(&dir);
        let writer =
            std::sync::Arc::new(ocserve_store::Writer::spawn(db.clone()).expect("temp writer"));
        let blobs = std::sync::Arc::new(
            ocserve_store::BlobStore::new(dir.join("blobs")).expect("temp blobs"),
        );
        Self::with_wiring(
            auth,
            p,
            Wires {
                db,
                blobs,
                writer,
                llm: LlmRegistry::default(),
            },
        )
    }

    /// One-shot construction with real runtime wiring (CLI path).
    pub fn with_wiring(auth: Option<(String, String)>, p: Payloads, w: Wires) -> Arc<Self> {
        let home = std::env::var("HOME").unwrap_or_default();
        let worktree = std::env::current_dir()
            .map(|p| p.display().to_string())
            .unwrap_or_else(|_| "/".into());
        let wire_off = std::env::var("OCSERVE_WIRE_CACHE")
            .map(|v| v == "0")
            .unwrap_or(false);
        // v2 envelope location (static per process — same value every request).
        let location = json!({
            "directory": worktree,
            "project": {"id": "global", "directory": "/"},
        });
        let wire_map = rebuild_wire(&p, wire_off, &location);
        let bus = ocserve_core::EventBus::new();
        let pty = {
            let mgr = ocserve_pty::PtyManager::new(worktree.clone());
            let bus_for_pty = bus.clone();
            let dir_for_pty = worktree.clone();
            mgr.set_event_sink(std::sync::Arc::new(move |event_type, properties| {
                bus_for_pty.publish(ocserve_core::event::frame(
                    &dir_for_pty,
                    event_type,
                    properties,
                ));
            }));
            mgr
        };
        Arc::new(Self {
            payloads: parking_lot::RwLock::new(p),
            wire: parking_lot::RwLock::new(wire_map),
            wire_off: std::sync::atomic::AtomicBool::new(wire_off),
            reloader: parking_lot::RwLock::new(None),
            // upstream /path shape (keys golden: home/state/config/worktree/directory)
            paths: json!({
                "home": home,
                "state": format!("{home}/.local/state/opencode"),
                "config": format!("{home}/.config/opencode"),
                "worktree": worktree,
                "directory": worktree,
            }),
            auth,
            requests: std::sync::atomic::AtomicU64::new(0),
            bus,
            gate: ocserve_core::PermissionGate::new(),
            mcp: std::sync::OnceLock::new(),
            plugins: std::sync::OnceLock::new(),
            prompt_locks: std::sync::Arc::new(parking_lot::Mutex::new(
                std::collections::HashMap::new(),
            )),
            prompt_tasks: std::sync::Arc::new(parking_lot::Mutex::new(
                std::collections::HashMap::new(),
            )),
            prompt_gen: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1)),
            question_gate: ocserve_core::question::QuestionGate::new(),
            tui: crate::tui::TuiControl::default(),
            ui: crate::ui::UiProxy::from_env(),
            pty,
            cors_extra: read_cors_env(),
            db: w.db,
            blobs: w.blobs,
            writer: w.writer,
            llm: parking_lot::RwLock::new(w.llm),
            watch: parking_lot::RwLock::new(watch::WatchState::default()),
            reconcile_lock: tokio::sync::Mutex::new(()),
            max_rounds: std::env::var("OCSERVE_PROMPT_MAX_ROUNDS")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(0),
            cost_ceiling: std::env::var("OCSERVE_PROMPT_MAX_COST_USD")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(0.0),
        })
    }
}

async fn health() -> impl IntoResponse {
    Json(json!({"healthy": true, "version": FREEZE_VERSION}))
}

/// `GET /doc` — serve the vendored OpenAPI contract (freeze parity).
async fn get_doc() -> impl IntoResponse {
    (
        [(axum::http::header::CONTENT_TYPE, "application/json")],
        crate::doc::SPEC_JSON,
    )
}

async fn get_config(
    State(st): State<Arc<AppState>>,
    headers: axum::http::HeaderMap,
) -> axum::response::Response {
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    wire_json_h(&st, "config", ae, inm).unwrap_or_else(|| {
        let v = st.payloads.read().config.clone();
        let body = serde_json::to_vec(&v).unwrap_or_default();
        let etag = crate::compress::etag_for(&body);
        crate::compress::encode_response(
            axum::http::StatusCode::OK,
            "application/json",
            bytes::Bytes::from(body),
            Some(&etag),
            ae,
            inm,
            &[],
        )
    })
}

async fn get_agent(
    State(st): State<Arc<AppState>>,
    headers: axum::http::HeaderMap,
) -> axum::response::Response {
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    wire_json_h(&st, "agent", ae, inm).unwrap_or_else(|| {
        let v = st.payloads.read().agent.clone();
        let body = serde_json::to_vec(&v).unwrap_or_default();
        let etag = crate::compress::etag_for(&body);
        crate::compress::encode_response(
            axum::http::StatusCode::OK,
            "application/json",
            bytes::Bytes::from(body),
            Some(&etag),
            ae,
            inm,
            &[],
        )
    })
}

async fn get_command(
    State(st): State<Arc<AppState>>,
    headers: axum::http::HeaderMap,
) -> axum::response::Response {
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    wire_json_h(&st, "command", ae, inm).unwrap_or_else(|| {
        let v = st.payloads.read().command.clone();
        let body = serde_json::to_vec(&v).unwrap_or_default();
        let etag = crate::compress::etag_for(&body);
        crate::compress::encode_response(
            axum::http::StatusCode::OK,
            "application/json",
            bytes::Bytes::from(body),
            Some(&etag),
            ae,
            inm,
            &[],
        )
    })
}

async fn get_config_providers(
    State(st): State<Arc<AppState>>,
    headers: axum::http::HeaderMap,
) -> axum::response::Response {
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    wire_json_h(&st, "config_providers", ae, inm).unwrap_or_else(|| {
        let v = st.payloads.read().config_providers.clone();
        let body = serde_json::to_vec(&v).unwrap_or_default();
        let etag = crate::compress::etag_for(&body);
        crate::compress::encode_response(
            axum::http::StatusCode::OK,
            "application/json",
            bytes::Bytes::from(body),
            Some(&etag),
            ae,
            inm,
            &[],
        )
    })
}

async fn get_provider(
    State(st): State<Arc<AppState>>,
    headers: axum::http::HeaderMap,
) -> axum::response::Response {
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    wire_json_h(&st, "provider", ae, inm).unwrap_or_else(|| {
        let v = st.payloads.read().provider.clone();
        let body = serde_json::to_vec(&v).unwrap_or_default();
        let etag = crate::compress::etag_for(&body);
        crate::compress::encode_response(
            axum::http::StatusCode::OK,
            "application/json",
            bytes::Bytes::from(body),
            Some(&etag),
            ae,
            inm,
            &[],
        )
    })
}

async fn get_console(
    State(st): State<Arc<AppState>>,
    headers: axum::http::HeaderMap,
) -> axum::response::Response {
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    wire_json_h(&st, "console", ae, inm).unwrap_or_else(|| {
        let v = st.payloads.read().console.clone();
        let body = serde_json::to_vec(&v).unwrap_or_default();
        let etag = crate::compress::etag_for(&body);
        crate::compress::encode_response(
            axum::http::StatusCode::OK,
            "application/json",
            bytes::Bytes::from(body),
            Some(&etag),
            ae,
            inm,
            &[],
        )
    })
}

async fn get_capabilities(
    State(st): State<Arc<AppState>>,
    headers: axum::http::HeaderMap,
) -> axum::response::Response {
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    wire_json_h(&st, "capabilities", ae, inm).unwrap_or_else(|| {
        let v = st.payloads.read().capabilities.clone();
        let body = serde_json::to_vec(&v).unwrap_or_default();
        let etag = crate::compress::etag_for(&body);
        crate::compress::encode_response(
            axum::http::StatusCode::OK,
            "application/json",
            bytes::Bytes::from(body),
            Some(&etag),
            ae,
            inm,
            &[],
        )
    })
}

/// v2-style location envelope shared by /api/* routes (captured live).
pub(crate) fn api_location(st: &AppState) -> Value {
    let dir = st.paths["directory"].as_str().unwrap_or("/");
    json!({
        "directory": dir,
        "project": {"id": "global", "directory": "/"},
    })
}

/// GET /api/agent — v2-shaped agent list (permission triple renamed to
/// {action, resource, effect}; natives-first order; no plugin agents).
async fn get_api_agent(
    State(st): State<Arc<AppState>>,
    headers: axum::http::HeaderMap,
) -> axum::response::Response {
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    crate::wire_json_h(&st, "api_agent", ae, inm).unwrap_or_else(|| {
        let payloads = st.payloads.read();
        Json(api_agent_body(&payloads, &api_location(&st))).into_response()
    })
}

/// Pure projection for /api/agent (cache + fallback share this).
pub(crate) fn api_agent_body(payloads: &Payloads, location: &Value) -> Value {
    json!({"location": location, "data": api_agent_data(payloads)})
}

fn api_agent_data(payloads: &Payloads) -> Vec<Value> {
    let data: Vec<Value> = payloads
        .api_agent
        .iter()
        .map(|a| {
            let perms: Vec<Value> = a["permission"]
                .as_array()
                .map(|rules| {
                    rules
                        .iter()
                        .map(|r| {
                            json!({
                                "action": r["permission"],
                                "resource": r["pattern"],
                                "effect": r["action"],
                            })
                        })
                        .collect()
                })
                .unwrap_or_default();
            let mut out = json!({
                "id": a["name"],
                "hidden": a.get("hidden").and_then(|v| v.as_bool()).unwrap_or(false),
                "mode": a["mode"],
                "permissions": perms,
                "request": {"headers": {}, "body": {}},
            });
            if let Some(d) = a.get("description") {
                out["description"] = d.clone();
            }
            // system: explicit prompt, else the captured build default
            if let Some(p) = a.get("prompt") {
                out["system"] = p.clone();
            } else if a["name"] == "build" {
                out["system"] = json!(BUILD_SYSTEM_BLURB);
            }
            out
        })
        .collect();
    data
}

/// Captured from live /api/agent (build has no prompt; upstream serves this blurb).
const BUILD_SYSTEM_BLURB: &str = "You are an AI coding agent. Help the user accomplish software \
engineering tasks by inspecting the workspace, making targeted changes, and using tools according \
to the configured permissions.";

/// GET /api/command — {location, data:[{description,name,template}]}.
async fn get_api_command(
    State(st): State<Arc<AppState>>,
    headers: axum::http::HeaderMap,
) -> axum::response::Response {
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    crate::wire_json_h(&st, "api_command", ae, inm).unwrap_or_else(|| {
        let payloads = st.payloads.read();
        Json(api_command_body(&payloads, &api_location(&st))).into_response()
    })
}

/// Pure projection for /api/command (cache + fallback share this).
pub(crate) fn api_command_body(payloads: &Payloads, location: &Value) -> Value {
    let data: Vec<Value> = payloads
        .command
        .iter()
        .map(|c| {
            json!({
                "name": c["name"],
                "description": c.get("description").and_then(|v| v.as_str()).unwrap_or(""),
                "template": c.get("template").and_then(|v| v.as_str()).unwrap_or(""),
            })
        })
        .collect();
    json!({"location": location, "data": data})
}

/// Empty-shape routes captured live: {}, [].
/// GET /experimental/tool?provider=&model= — v1 ToolList (groups/
/// experimental.ts:50-61; handlers/experimental.ts:94-105): [{id,
/// description, parameters}] = builtin + MCP tools. Effect validates the
/// query (missing provider/model → BadRequest) — mirrored here. Denied
/// tools excluded, matching what the model actually sees (W5).
async fn get_experimental_tool(
    State(st): State<Arc<AppState>>,
    axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Value>, ApiError> {
    let has = |k: &str| q.get(k).map(|s| !s.is_empty()).unwrap_or(false);
    if !has("provider") || !has("model") {
        return Err(ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: "Invalid query: provider, model required".into(),
        });
    }
    let mut schemas = ocserve_tools::schemas();
    if let Some(hub) = st.mcp.get() {
        let mcp_tools =
            tokio::time::timeout(std::time::Duration::from_secs(10), hub.tool_schemas())
                .await
                .unwrap_or_else(|_| {
                    tracing::warn!("mcp tool schema fetch timed out (/experimental/tool)");
                    Vec::new()
                });
        schemas.extend(mcp_tools);
    }
    let items: Vec<Value> = schemas
        .iter()
        .filter_map(|t| {
            let f = t.get("function")?;
            Some(json!({
                "id": f.get("name")?,
                "description": f.get("description")?,
                "parameters": f.get("parameters")?,
            }))
        })
        .collect();
    Ok(Json(Value::Array(items)))
}

/// GET /experimental/tool/ids — registry.ids(): plain string array.
async fn get_experimental_tool_ids() -> Json<Value> {
    let ids: Vec<Value> = ocserve_tools::schemas()
        .iter()
        .filter_map(|t| t.pointer("/function/name").and_then(|n| n.as_str()))
        .map(|n| json!(n))
        .collect();
    Json(Value::Array(ids))
}

async fn experimental_resource() -> Json<Value> {
    Json(json!({}))
}
async fn experimental_workspace() -> Json<Value> {
    Json(json!([]))
}

/// GET /experimental/workspace/status — freeze: `[]` 200 (probed live).
async fn experimental_workspace_status() -> Json<Value> {
    Json(json!([]))
}
async fn formatter_list() -> Json<Value> {
    Json(json!([]))
}
async fn lsp_list() -> Json<Value> {
    Json(json!([]))
}
async fn project_directories() -> Json<Value> {
    Json(json!([]))
}

/// GET /mcp — {name: {status, error?}} (freeze capture §1146).
async fn get_mcp(State(st): State<Arc<AppState>>) -> Json<Value> {
    let hub = st
        .mcp
        .get()
        .cloned()
        .unwrap_or_else(|| std::sync::Arc::new(ocserve_mcp::McpHub::default()));
    Json(hub.statuses())
}

/// POST /mcp (v1 `mcp.add`) — dynamically register an MCP server, then return
/// the full status map (freeze probed 2026-10-09: `{name:{status,error?}}`).
/// Missing `name`/`config` → 400 `{"name":"BadRequest","data":{message,kind}}`.
/// The server is added via the same hub path the config hot-reload uses
/// (upsert_cfg + connect), so a subsequently-listed server is live.
async fn post_mcp(
    State(st): State<Arc<AppState>>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, HttpError> {
    let name = body
        .get("name")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
        .ok_or_else(|| mcp_payload_err("name"))?;
    let config = body
        .get("config")
        .filter(|c| c.is_object())
        .ok_or_else(|| mcp_payload_err("config"))?;
    let hub = st.mcp.get().cloned().ok_or_else(|| {
        HttpError::Api(ApiError {
            status: StatusCode::SERVICE_UNAVAILABLE,
            name: "McpUnavailable",
            message: "MCP hub not initialized".into(),
        })
    })?;
    // Build a single-entry `mcp` map and reuse the exact config parser so the
    // POST shape and the config-file shape can never diverge.
    let mut one = serde_json::Map::new();
    one.insert(name.to_string(), config.clone());
    let cfg = ocserve_mcp::parse_config(&Value::Object(one))
        .into_iter()
        .next()
        .ok_or_else(|| mcp_payload_err("config"))?;
    hub.upsert_cfg(cfg);
    // connect failure is recorded in the status map (freeze returns the map
    // with a `failed` entry, HTTP 200) — never a 500 for an unreachable server.
    if let Err(e) = hub.connect(name).await {
        tracing::info!("mcp add: connect {name} failed (recorded): {e:#}");
    }
    Ok(Json(hub.statuses()))
}

/// Freeze `McpPaths.status` payload error: `Missing key at ["<field>"]`.
fn mcp_payload_err(field: &str) -> HttpError {
    HttpError::Payload {
        message: format!("Missing key\n  at [\"{field}\"]"),
    }
}

/// Boundary wire-decode for v1 routes: freeze renders decode failures as
/// `{"name":"BadRequest","data":{"message":…,"kind":"Payload"}}` [400]
/// (field-probes.md). Decode runs FIRST — before session-existence checks
/// (probe: `{}` on a nonexistent session → 400, not 404).
pub(crate) fn decode_v1<T>(r: Result<T, ocserve_core::wire::WireError>) -> Result<T, HttpError> {
    r.map_err(|e| HttpError::Payload { message: e.message })
}

/// Boundary wire-decode for v2 `/api/*` routes: freeze renders
/// `{"_tag":"InvalidRequestError","message":…,"kind":"Payload"}` [400]
/// (field-probes.md — differs from v1 on BOTH tag and nesting).
pub(crate) fn decode_v2<T>(r: Result<T, ocserve_core::wire::WireError>) -> Result<T, HttpError> {
    r.map_err(|e| HttpError::TaggedData {
        status: StatusCode::BAD_REQUEST,
        tag: "InvalidRequestError",
        fields: json!({"message": e.message, "kind": "Payload"}),
    })
}

/// GET /experimental/session?search=&roots=&limit= — session title search
/// (oc-remote searchSessions: "not a content search"); rows = list wire shape
/// + embedded project {id, worktree} (observed contract, manifest keys mode).
async fn get_experimental_sessions(
    State(st): State<Arc<AppState>>,
    axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Json<Value> {
    let mut v = run_blocking(&st.db, ocserve_store::load_sessions_wire)
        .await
        .unwrap_or_else(|e| {
            tracing::error!("session list read failed: {e:#}");
            Vec::new()
        });
    if let Some(search) = q.get("search").filter(|s| !s.is_empty()) {
        let needle = search.to_lowercase();
        v.retain(|s| {
            s["title"]
                .as_str()
                .unwrap_or_default()
                .to_lowercase()
                .contains(&needle)
        });
    }
    if let Some(limit) = q.get("limit").and_then(|l| l.parse::<usize>().ok()) {
        v.truncate(limit);
    }
    // roots=true is implicit: our store holds root sessions only (M3 scope).
    for row in &mut v {
        row["project"] = json!({"id": "global", "worktree": "/"});
    }
    Json(Value::Array(v))
}

/// POST /permission/{id}/reply — {reply: once|always|reject} (oc-remote contract).
/// Body decode precedes the 404 (freeze probe: `{}` → 400 Payload "Missing key
/// at [reply]"; `{"reply":"maybe"}`/`{"reply":123}` → 400 Payload "Expected
/// once | always | reject"). The old `unwrap_or("reject")` silently coerced a
/// malformed body into a REJECT of a real pending ask and returned 404 for
/// unknown ids instead of the decode 400.
async fn post_permission_reply(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, HttpError> {
    let reply = validate_permission_reply(&body)?;
    if !st.gate.reply(&id, reply) {
        return Err(HttpError::TaggedData {
            status: StatusCode::NOT_FOUND,
            tag: "PermissionNotFoundError",
            fields: json!({"requestID": id, "message": format!("Permission request not found: {id}")}),
        });
    }
    Ok(Json(json!({})))
}

/// Validate a permission reply enum with the freeze Payload-decode error
/// messages (order matters: missing key first, then the union value).
pub(crate) fn validate_permission_reply(body: &Value) -> Result<&'static str, HttpError> {
    validate_permission_reply_key(body, "reply")
}

/// Same, for the deprecated `{response: …}` body key.
pub(crate) fn validate_permission_reply_key(
    body: &Value,
    key: &str,
) -> Result<&'static str, HttpError> {
    let Some(v) = body.get(key) else {
        return Err(HttpError::Payload {
            message: format!("Missing key\n  at [\"{key}\"]"),
        });
    };
    match v.as_str() {
        Some("once") => Ok("once"),
        Some("always") => Ok("always"),
        Some("reject") => Ok("reject"),
        // freeze renders the offending value with JSON syntax (strings quoted,
        // numbers bare): `Expected "once" | "always" | "reject", got 123`.
        _ => Err(HttpError::Payload {
            message: format!(
                "Expected \"once\" | \"always\" | \"reject\", got {}\n  at [\"{key}\"]",
                v
            ),
        }),
    }
}

/// GET /permission — pending permission requests.
async fn get_permissions(State(st): State<Arc<AppState>>) -> Json<Value> {
    Json(serde_json::Value::Array(st.gate.list()))
}

/// GET /vcs — {branch, default_branch} (values cwd-dependent; keys golden).
async fn vcs_info(State(st): State<Arc<AppState>>) -> impl IntoResponse {
    let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
    let info = run_blocking(&st.db, move |_db| Ok::<_, anyhow::Error>(vcs::info(&dir)))
        .await
        .unwrap_or_else(|_| json!({"branch": null, "default_branch": null}));
    Json(info)
}

/// GET /vcs/status → FileStatus[] (freeze shape; [] when not a git repo).
async fn vcs_status(State(st): State<Arc<AppState>>) -> Response {
    let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
    match run_blocking(&st.db, move |_db| Ok::<_, anyhow::Error>(vcs::status(&dir))).await {
        Ok(v) => axum::Json(v).into_response(),
        Err(_) => axum::Json(Value::Array(vec![])).into_response(),
    }
}

/// GET /vcs/diff?mode=git|branch[&context=N] → FileDiff[].
/// Missing/other mode → 400 Effect Query envelope (freeze probe).
async fn vcs_diff(
    State(st): State<Arc<AppState>>,
    axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
    headers: axum::http::HeaderMap,
) -> Response {
    let mode = q.get("mode").map(String::as_str).unwrap_or("");
    if mode != "git" && mode != "branch" {
        return HttpError::Query {
            message: format!("Expected \"git\" | \"branch\", got \"{mode}\"\n  at [\"mode\"]"),
        }
        .into_response();
    }
    let context: Option<u32> = q.get("context").and_then(|v| v.parse().ok());
    let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
    let mode_owned = mode.to_string();
    let v = run_blocking(&st.db, move |_db| {
        Ok::<_, anyhow::Error>(vcs::diff(&dir, &mode_owned, context))
    })
    .await
    .unwrap_or_else(|_| Value::Array(vec![]));
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    let body = serde_json::to_vec(&v).unwrap_or_default();
    let etag = crate::compress::etag_for(&body);
    crate::compress::encode_response(
        StatusCode::OK,
        "application/json",
        bytes::Bytes::from(body),
        Some(&etag),
        ae,
        inm,
        &[],
    )
}

/// GET /vcs/diff/raw → raw patch text (empty string when clean), bounded by the
/// same 10 MB cap the tracked diff uses (was unbounded — a large dirty tree
/// allocated the whole patch on a request).
async fn vcs_diff_raw(State(st): State<Arc<AppState>>) -> Response {
    let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
    let (text, truncated) = run_blocking(&st.db, move |_db| {
        Ok::<_, anyhow::Error>(vcs::diff_raw_bounded(&dir))
    })
    .await
    .unwrap_or_else(|_| (String::new(), false));
    Response::builder()
        .header(
            axum::http::header::CONTENT_TYPE,
            "text/plain; charset=utf-8",
        )
        .header(axum::http::header::VARY, "Accept-Encoding")
        .header("x-ocserve-truncated", if truncated { "1" } else { "0" })
        .body(axum::body::Body::from(text))
        .expect("static response")
}

/// GET /file/status → [] (freeze returns the changed-file set from the
/// snapshot engine; with no snapshot engine the honest shape is empty —
/// the web UI only checks length>0 for a dirty badge, which `vcs/status`
/// now serves correctly).
async fn file_status() -> impl IntoResponse {
    axum::Json(Value::Array(vec![]))
}

/// GET /find/symbol → [] (freeze returns LSP symbols; no LSP server →
/// empty, the same shape clients tolerate).
async fn find_symbol() -> impl IntoResponse {
    axum::Json(Value::Array(vec![]))
}

/// GET /global/config → the effective merged config (freeze shape: same
/// keys as /config minus the per-instance `agent`/`command`/`mode`/`username`
/// overrides — probed key diff 2026-10-08). Served from the wire cache.
async fn global_config_get(
    State(st): State<Arc<AppState>>,
    headers: axum::http::HeaderMap,
) -> Response {
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    // Freeze /global/config == /config with the instance-scoped keys absent.
    // Build it once per reload into the wire cache under "global_config".
    wire_json_h(&st, "global_config", ae, inm).unwrap_or_else(|| {
        let mut v = st.payloads.read().config.clone();
        if let Some(obj) = v.as_object_mut() {
            for k in ["agent", "command", "mode", "username"] {
                obj.remove(k);
            }
        }
        let body = serde_json::to_vec(&v).unwrap_or_default();
        let etag = crate::compress::etag_for(&body);
        crate::compress::encode_response(
            StatusCode::OK,
            "application/json",
            bytes::Bytes::from(body),
            Some(&etag),
            ae,
            inm,
            &[],
        )
    })
}

// ---- PERF-10X F9: paged /message wire memo ----
//
// message_page was the CPU hog (route isolation: ~3 ms CPU/page — parse +
// merge + serialize per message, blocking-pool thread, channel framing).
// A paged response (limit>0, ≤50 messages) is assembled ONCE per
// (db, session, limit, cursor, write-epoch) and served as refcounted
// Bytes; hits skip reader, queries, parse, thread and channel entirely.
// Bound by construction: entries ≤ 4 MiB (larger assemblies abort and
// fall back to the original streaming path — the unbounded full-history
// stream is NEVER memoized, AGENTS §2.3), FIFO capped at 32 entries.
// Exactness: keys carry the write epoch; a writer bump invalidates
// (lazy clear on epoch mismatch — same rule as F7/F8). Kill switch:
// OCSERVE_PAGE_MEMO=0.
pub const PAGE_MEMO_ENTRY_CAP: usize = 4 * 1024 * 1024;
/// Count + BYTE bounds (F9c): the first cap (32) thrashed against the
/// fixture's 200-session spread working set (VU150/300 spread fell to
/// 3,794 rps, p95 278ms — every rotation missed and paid the full
/// assemble path; arrival never recovered from it). Working set ≈
/// sessions x pages-per-loop ≈ 600 keys; typical pages are 10-50KB so a
/// 64 MiB total budget holds the whole set (env: OCSERVE_PAGE_MEMO_MB).
pub const PAGE_MEMO_MAX_ENTRIES: usize = 2048;
pub const PAGE_MEMO_MAX_BYTES: usize = 64 * 1024 * 1024;

struct PageEntry {
    body: bytes::Bytes,
    next: Option<String>,
}

struct PageMemoState {
    epoch: u64,
    entries: std::collections::VecDeque<(String, PageEntry)>,
    bytes: usize,
}

impl PageMemoState {
    fn new() -> Self {
        Self {
            epoch: 0,
            entries: std::collections::VecDeque::new(),
            bytes: 0,
        }
    }
    fn get(&mut self, key: &str, epoch: u64) -> Option<PageEntry> {
        if self.epoch != epoch {
            self.entries.clear();
            self.bytes = 0;
            self.epoch = epoch;
        }
        self.entries
            .iter()
            .find(|(k, _)| k == key)
            .map(|(_, e)| PageEntry {
                body: e.body.clone(),
                next: e.next.clone(),
            })
    }
    fn put(&mut self, key: String, entry: PageEntry, epoch: u64) {
        if entry.body.len() > PAGE_MEMO_ENTRY_CAP.min(PAGE_MEMO_MAX_BYTES) {
            return;
        }
        if self.epoch != epoch {
            self.entries.clear();
            self.bytes = 0;
            self.epoch = epoch;
        }
        // drop same-key old copy (byte accounting)
        if let Some(pos) = self.entries.iter().position(|(k, _)| k == &key) {
            let (_, old) = self.entries.remove(pos).expect("position found");
            self.bytes -= old.body.len();
        }
        // RAM audit 2026-10-11: the env budget was unbounded above — a large
        // OCSERVE_PAGE_MEMO_MB could hold arbitrarily many bytes, and a single
        // entry larger than the budget was kept alone (len() == 1 skips the
        // eviction loop, so `bytes > budget` with no victim). Clamp the
        // override to the const ceiling and refuse entries that alone exceed
        // it (entry cap above already bounds each, so a refused entry is
        // simply not memoized — the route still serves from the DB).
        let budget = std::env::var("OCSERVE_PAGE_MEMO_MB")
            .ok()
            .and_then(|v| v.parse::<usize>().ok())
            .map(|mb| mb * 1024 * 1024)
            .unwrap_or(PAGE_MEMO_MAX_BYTES)
            .min(PAGE_MEMO_MAX_BYTES);
        self.bytes += entry.body.len();
        self.entries.push_back((key, entry));
        while self.entries.len() > PAGE_MEMO_MAX_ENTRIES
            || (self.bytes > budget && self.entries.len() > 1)
        {
            if let Some((_, old)) = self.entries.pop_front() {
                self.bytes -= old.body.len();
            } else {
                break;
            }
        }
    }
    fn len(&self) -> usize {
        self.entries.len()
    }
}

static PAGE_MEMO: std::sync::OnceLock<parking_lot::Mutex<PageMemoState>> =
    std::sync::OnceLock::new();
static PAGE_MEMO_HITS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
static PAGE_MEMO_MISSES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

fn page_memo() -> &'static parking_lot::Mutex<PageMemoState> {
    PAGE_MEMO.get_or_init(|| parking_lot::Mutex::new(PageMemoState::new()))
}

pub fn page_memo_enabled() -> bool {
    std::env::var("OCSERVE_PAGE_MEMO")
        .map(|v| v != "0")
        .unwrap_or(true)
}

/// (hits, misses, entries) for the load report / tests.
pub fn page_memo_stats() -> (u64, u64, usize) {
    (
        PAGE_MEMO_HITS.load(std::sync::atomic::Ordering::Relaxed),
        PAGE_MEMO_MISSES.load(std::sync::atomic::Ordering::Relaxed),
        page_memo().lock().len(),
    )
}

/// Page response builder shared by the memo path and the streaming path
/// (headers rule = upstream session.ts:133-147: Link echoes the request
/// origin — Host / x-forwarded-proto — so headers are rebuilt per request
/// even on a memo hit; only the BODY is cached).
fn page_response(
    headers: &axum::http::HeaderMap,
    uri: &axum::http::Uri,
    page_n: &Option<u64>,
    page_next: &Option<String>,
    body: axum::body::Body,
) -> Result<axum::response::Response, ApiError> {
    let mut builder = axum::response::Response::builder()
        .header(axum::http::header::CONTENT_TYPE, "application/json");
    if let (Some(n), Some(cur)) = (page_n, page_next) {
        let scheme = headers
            .get("x-forwarded-proto")
            .and_then(|v| v.to_str().ok())
            .unwrap_or("http");
        let host = headers
            .get("host")
            .and_then(|v| v.to_str().ok())
            .unwrap_or("localhost");
        let link = format!(
            "<{scheme}://{host}{}?limit={n}&before={cur}>; rel=\"next\"",
            uri.path()
        );
        builder = builder
            .header("x-next-cursor", cur.as_str())
            .header("link", link)
            .header("access-control-expose-headers", "Link, X-Next-Cursor");
    }
    builder.body(body).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("response build: {e}"),
    })
}

async fn get_sessions(
    State(st): State<Arc<AppState>>,
    axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
    headers: axum::http::HeaderMap,
) -> axum::response::Response {
    // W4: freeze ListQuery parity (limit default 100, roots, search, start,
    // directory, scope=project). The bare list keeps the exact single-slot
    // memo path (byte-identical); filtered variants use a param-keyed memo.
    // Fresh DB rows (upstream semantics) — the boot map was stale after
    // prompts (M2 finding); store orders by time_updated DESC.
    let query = ocserve_store::SessionListQuery {
        limit: q
            .get("limit")
            .and_then(|v| v.parse::<usize>().ok())
            .filter(|n| *n > 0),
        roots: q.get("roots").map(|v| v == "true").unwrap_or(false),
        search: q.get("search").filter(|s| !s.is_empty()).cloned(),
        start: q.get("start").and_then(|v| v.parse::<i64>().ok()),
        directory: q.get("directory").filter(|s| !s.is_empty()).cloned(),
        scope_project: q.get("scope").map(|v| v == "project").unwrap_or(false),
    };
    let ae = headers
        .get(axum::http::header::ACCEPT_ENCODING)
        .and_then(|v| v.to_str().ok());
    let inm = headers
        .get(axum::http::header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    let qc = query.clone();
    match run_blocking(&st.db, move |db| {
        ocserve_store::load_sessions_wire_bytes_filtered(db, &qc)
    })
    .await
    {
        Ok(b) => {
            let etag = crate::compress::etag_for(&b);
            crate::compress::encode_response(
                axum::http::StatusCode::OK,
                "application/json",
                (*b).clone(),
                Some(&etag),
                ae,
                inm,
                &[],
            )
        }
        Err(e) => {
            tracing::error!("session list read failed: {e:#}");
            Json(Vec::<serde_json::Value>::new()).into_response()
        }
    }
}

async fn get_session(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<impl IntoResponse, ApiError> {
    let wire_id = id.clone();
    run_blocking(&st.db, move |db| {
        ocserve_store::load_session_wire(db, &wire_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })?
    .map(Json)
    .ok_or_else(|| ApiError::not_found(format!("Session not found: {id}")))
}

/// GET /session/status — map of NON-idle sessions only (upstream
/// session/status.ts deletes idle entries; probe fixture: status_busy.body).
/// Busy = held/contended prompt lock OR a live background prompt task —
/// queued-but-not-yet-running prompts count (client's send is admitted).
async fn session_status(State(st): State<Arc<AppState>>) -> impl IntoResponse {
    let mut out = serde_json::Map::new();
    let mut busy = |sid: &str| {
        out.insert(sid.to_string(), serde_json::json!({"type": "busy"}));
    };
    for (sid, arc) in st.prompt_locks.lock().iter() {
        if arc.try_lock().is_err() {
            busy(sid);
        }
    }
    for (sid, (_gen, handle)) in st.prompt_tasks.lock().iter() {
        if !handle.is_finished() {
            busy(sid);
        }
    }
    Json(serde_json::Value::Object(out))
}

/// GET /path — env-derived at boot (upstream keys: home/state/config/worktree/directory).
async fn get_path(State(st): State<Arc<AppState>>) -> impl IntoResponse {
    Json(st.paths.clone())
}

/// GET /project — the global project + one project per distinct session
/// directory (newest first). freeze derives its project list from the worktree
/// registry; ocserve derives it from stored session rows (documented
/// derivation divergence, same wire shape). This is what lets the web UI group
/// sessions under the directories they actually belong to — with only
/// `global`+`current`, sessions under other directories were dropped from the
/// sidebar ("Nothing here yet").
async fn get_projects(State(st): State<Arc<AppState>>) -> impl IntoResponse {
    let now_ms = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0);
    let mut dirs = ocserve_store::distinct_session_directories(&st.db).unwrap_or_default();
    // ensure the server worktree is always present even on an empty/fresh DB
    let worktree = st.paths["worktree"].as_str().unwrap_or("/").to_string();
    if !dirs.iter().any(|d| d == &worktree) {
        dirs.insert(0, worktree);
    }
    let project = |id: &str, wt: &str| {
        json!({
            "id": id,
            "worktree": wt,
            "time": {"created": now_ms, "updated": now_ms, "initialized": now_ms},
            "sandboxes": [],
        })
    };
    let mut out = vec![project("global", "/")];
    for d in dirs {
        out.push(project(&d, &d));
    }
    Json(Value::Array(out))
}

/// GET /project/current — same shape, single object.
async fn get_project_current(State(st): State<Arc<AppState>>) -> impl IntoResponse {
    let now_ms = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0);
    let worktree = st.paths["worktree"].as_str().unwrap_or("/");
    Json(json!({
        "id": "global",
        "worktree": worktree,
        "time": {"created": now_ms, "updated": now_ms, "initialized": now_ms},
        "sandboxes": [],
    }))
}

/// Prometheus text format for the KPI series (SRE §2). Internal bind only.
async fn metrics(State(_st): State<Arc<AppState>>) -> impl IntoResponse {
    // registry counters are incremented by the timing middleware; rss sampled
    // here (scrape-time) so gauges are fresh
    ocserve_metrics::sample_rss();
    // Read-path fast activity (L1 splice + F5/F7/F8 memos), published as
    // absolute cumulative gauges at scrape time — these counters live in
    // ocserve-store as atomics with no event stream of their own.
    //  - splice_rows/fallbacks: the corpus gate proves 0 fallbacks on real
    //    data, so a non-zero fallback gauge means stored bytes hit a shape
    //    the splicer refuses — a signal, not a silent fallback (lib.rs L1).
    //  - memo hit/miss: the load report's cache-on/off variants must be
    //    observable in production, not just in a bench run.
    ocserve_metrics::gauge(
        "ocserve_splice_rows_total",
        ocserve_store::splice_rows() as i64,
    );
    ocserve_metrics::gauge(
        "ocserve_splice_fallbacks_total",
        ocserve_store::splice_fallbacks() as i64,
    );
    let (search_hits, search_misses, list_hits, list_misses, _pad, search_len, _epoch) =
        ocserve_store::memo_stats();
    ocserve_metrics::gauge("ocserve_memo_search_hits", search_hits as i64);
    ocserve_metrics::gauge("ocserve_memo_search_misses", search_misses as i64);
    ocserve_metrics::gauge("ocserve_memo_list_hits", list_hits as i64);
    ocserve_metrics::gauge("ocserve_memo_list_misses", list_misses as i64);
    ocserve_metrics::gauge("ocserve_memo_search_entries", search_len as i64);
    let body = ocserve_metrics::render();
    (
        [(
            axum::http::header::CONTENT_TYPE,
            "text/plain; version=0.0.4",
        )],
        body,
    )
}

// ---- M2: session create + sync prompt + message list ----

/// Two-word slug (upstream generates like "stellar-orchid" — entropy from id).
fn slug_for(id: &str) -> String {
    const A: &[&str] = &[
        "stellar", "amber", "quiet", "cobalt", "rapid", "lunar", "vivid", "crisp", "amber", "ivory",
    ];
    const B: &[&str] = &[
        "orchid", "falcon", "meadow", "canyon", "ember", "drift", "quartz", "harbor", "summit",
        "ripple",
    ];
    let h = id
        .bytes()
        .fold(0u64, |acc, b| acc.wrapping_mul(31).wrapping_add(b as u64));
    format!(
        "{}-{}",
        A[(h % A.len() as u64) as usize],
        B[((h >> 16) % B.len() as u64) as usize]
    )
}

fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

/// ISO-8601 UTC with milliseconds (freeze `Date.prototype.toISOString`
/// shape) — days-from-civil, no time-crate dependency.
fn iso8601_z(ms: i64) -> String {
    let ms = ms.rem_euclid(86_400_000_000_000);
    let (days, rem) = (ms.div_euclid(86_400_000), ms.rem_euclid(86_400_000));
    let (h, rem) = (rem.div_euclid(3_600_000), rem.rem_euclid(3_600_000));
    let (mi, rem) = (rem.div_euclid(60_000), rem.rem_euclid(60_000));
    let (sec, milli) = (rem.div_euclid(1_000), rem.rem_euclid(1_000));
    // civil_from_days (Howard Hinnant)
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z.rem_euclid(146_097);
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    format!("{y:04}-{m:02}-{d:02}T{h:02}:{mi:02}:{sec:02}.{milli:03}Z")
}

/// Freeze default session title (session.ts `parentTitlePrefix + ISO`) —
/// JS `Date.prototype.toDateString()` ("Mon Oct 12 2026") — the env block's
/// "Today's date" line (system.ts:environment). Days-from-civil, no date crate.
pub fn js_to_date_string(ms: i64) -> String {
    const WDAY: [&str; 7] = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"];
    const MON: [&str; 12] = [
        "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec",
    ];
    let days = ms.div_euclid(86_400_000);
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z.rem_euclid(146_097);
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 } as usize;
    let y = if m <= 2 { y + 1 } else { y };
    // 1970-01-01 was a Thursday (weekday 4; JS Sun=0)
    let wd = (days + 4).rem_euclid(7) as usize;
    format!("{} {} {:02} {}", WDAY[wd], MON[m - 1], d, y)
}

fn default_session_title(now_ms: i64) -> String {
    format!("New session - {}", iso8601_z(now_ms))
}

/// Freeze `getForkedTitle` (session.ts): `/^(.+) \(fork #(\d+)\)$/` →
/// `base (fork #(n+1))`, otherwise `title (fork #1)`. Ported with string ops
/// (regex `.+` needs ≥1 char before the suffix → `idx > 0`, greedy match →
/// last occurrence → `rfind`).
pub fn forked_title(title: &str) -> String {
    const PAT: &str = " (fork #";
    if let Some(idx) = title.rfind(PAT)
        && idx > 0
        && let Some(num) = title[idx + PAT.len()..].strip_suffix(')')
        && !num.is_empty()
        && num.bytes().all(|b| b.is_ascii_digit())
        && let Ok(n) = num.parse::<u64>()
    {
        return format!("{} (fork #{})", &title[..idx], n + 1);
    }
    format!("{title} (fork #1)")
}

/// P0e: forward bus frames to the plugin sidecar's `event` hook (v1
/// plugin/index.ts:255-259). Lagged frames are skipped (upstream is
/// fire-and-forget); a closed bus ends the pump. Never blocks publishers.
pub fn start_plugin_event_pump(st: &Arc<AppState>) {
    // Subscribe SYNCHRONOUSLY here: last_seq is captured at call time, so
    // frames published between call and first recv are buffered (drain), not
    // skipped. (Spawning first raced: sync route handlers can publish before
    // the task ever ran — caught by the pump test.)
    let mut rx = st.bus.subscribe();
    let st = st.clone();
    tokio::spawn(async move {
        loop {
            match rx.recv().await {
                Ok(frame_str) => {
                    let Some(plug) = st.plugins.get() else {
                        continue; // sidecar not loaded — frames still on SSE
                    };
                    let Ok(v) = serde_json::from_str::<Value>(&frame_str) else {
                        continue;
                    };
                    let Some(payload) = v.get("payload").cloned() else {
                        continue;
                    };
                    let mut guard = plug.lock().await;
                    if let Err(e) = guard.emit_event(payload).await {
                        tracing::debug!("plugin event pump: {e:#}");
                    }
                }
                Err(ocserve_core::event::RecvError::Lagged(n)) => {
                    tracing::warn!("plugin event pump lagged {n} frames (skipped)");
                }
                Err(ocserve_core::event::RecvError::Closed) => break,
            }
        }
    });
}

/// POST /session — create (wire shape from captured fixture: cost, directory,
/// id, path, projectID, slug, time, title, tokens, version).
async fn post_session(
    State(st): State<Arc<AppState>>,
    axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
    Json(body): Json<Value>,
) -> Result<impl IntoResponse, ApiError> {
    let resolved =
        resolve_create_dir_path(&st, &body, q.get("directory").map(String::as_str)).await;
    Ok(Json(create_session_record(&st, &body, resolved)?))
}

/// Async directory resolution: the git toplevel lookup spawns a subprocess, so
/// it runs on the blocking pool (never pins an async worker — the M3 class).
async fn resolve_create_dir_path(
    st: &Arc<AppState>,
    body: &Value,
    query_dir: Option<&str>,
) -> (String, String) {
    let requested = resolve_create_dir(st, body, query_dir);
    let st = st.clone();
    tokio::task::spawn_blocking(move || freeze_dir_path(&requested))
        .await
        .unwrap_or_else(|_| {
            let d = resolve_create_dir(&st, body, query_dir);
            (d.clone(), d.trim_start_matches('/').to_string())
        })
}

/// Resolve the create-time directory (freeze semantics): `directory` query
/// param (v1) wins, else `body.location.directory` (v2 create shape), else the
/// server worktree. Before this, ocserve ignored the requested directory and
/// stamped every session with the server cwd — the "session thinks it's in the
/// ocserve repo" + constant `external_directory` prompt class.
fn resolve_create_dir(st: &Arc<AppState>, body: &Value, query_dir: Option<&str>) -> String {
    query_dir
        .filter(|s| !s.is_empty())
        .map(String::from)
        .or_else(|| {
            body.pointer("/location/directory")
                .and_then(|d| d.as_str())
                .filter(|s| !s.is_empty())
                .map(String::from)
        })
        .unwrap_or_else(|| st.paths["worktree"].as_str().unwrap_or("/").to_string())
}

/// freeze directory semantics (probed): if `dir` is inside a git repo, the
/// project `directory` is the repo toplevel and `path` is `dir` relative to it
/// (empty when equal, no leading slash); if it is NOT in a repo, `directory` is
/// `dir` itself and `path` is `dir` with the leading slash stripped (the cwd
/// relative to the global "/" project root).
/// Probed: a repo checkout → (toplevel, ""); a subdir of a repo →
/// (toplevel, "sub/dir"); a non-repo dir `/tmp/xyz` → ("/tmp/xyz", "tmp/xyz").
fn freeze_dir_path(dir: &str) -> (String, String) {
    let dir = dir.trim_end_matches('/');
    // `/` (or empty) is the global project root: no repo lookup — running
    // `git -C ""` would silently default to the SERVER cwd and mis-stamp the
    // session with the ocserve repo (the `?directory=/` env-block bug).
    if dir.is_empty() {
        return ("/".to_string(), String::new());
    }
    match git_toplevel(dir) {
        Some(root) => {
            let path = dir
                .strip_prefix(&root)
                .unwrap_or("")
                .trim_start_matches('/')
                .to_string();
            (root, path)
        }
        None => {
            let path = dir.trim_start_matches('/').to_string();
            (dir.to_string(), path)
        }
    }
}

/// `git rev-parse --show-toplevel` for `dir` (no shell). None when the dir is
/// absent, not a repo, or git is unavailable.
fn git_toplevel(dir: &str) -> Option<String> {
    let out = std::process::Command::new("git")
        .arg("-C")
        .arg(dir)
        .args(["rev-parse", "--show-toplevel"])
        .output()
        .ok()?;
    if !out.status.success() {
        return None;
    }
    let root = String::from_utf8_lossy(&out.stdout).trim().to_string();
    (!root.is_empty()).then_some(root)
}

/// Shared session creation — `POST /session` (v1) and `POST /api/session`
/// (v2) both land here. `worktree`/`rel_path` are the freeze-resolved project
/// root + cwd-relative path (see `freeze_dir_path`). Returns the created
/// Session.Info.
pub(crate) fn create_session_record(
    st: &Arc<AppState>,
    body: &Value,
    (worktree, rel_path): (String, String),
) -> Result<Value, ApiError> {
    let id = ocserve_core::ids::ses_id();
    let now = now_ms();
    let title = body
        .get("title")
        .and_then(|t| t.as_str())
        .unwrap_or("")
        .to_string();
    // K-TITLE: freeze default when the client sends none/empty — an empty
    // title renders as "untitled" everywhere (upstream: parentTitlePrefix).
    let title = if title.trim().is_empty() {
        default_session_title(now)
    } else {
        title
    };
    let agent = body
        .get("agent")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
        .unwrap_or("build");
    let model = body.get("model").cloned().unwrap_or(Value::Null);
    let mut info = json!({
        "id": id,
        "projectID": "global",
        "directory": worktree,
        "path": rel_path,
        "slug": slug_for(&id),
        "title": title,
        "version": FREEZE_VERSION,
        "time": {"created": now, "updated": now},
        "cost": 0,
        "tokens": {"input": 0, "output": 0, "reasoning": 0,
                   "cache": {"read": 0, "write": 0}},
    });
    if body.get("agent").is_some() {
        info["agent"] = json!(agent);
    }
    if model.is_object() {
        info["model"] = model;
    }
    ocserve_store::insert_session(&st.writer, &info).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })?;
    st.bus.publish(ocserve_core::event::frame(
        &worktree,
        "session.created",
        json!({"sessionID": id, "info": info}),
    ));
    Ok(info)
}

/// POST /session/{id}/fork — freeze session.fork (groups/session.ts:248,
/// session.ts:691). Effect decodes the payload BEFORE the handler runs, so
/// payload errors precede the 404: bad JSON / non-object / non-string
/// `messageID` → 400 `{"_tag":"BadRequest"}`; empty or whitespace body is
/// the NoContent arm (copy all). Returns the NEW Session.Info; messages are
/// copied strictly before `messageID` (freeze `slice(0, findIndex)`; unknown
/// id → findIndex -1 → copy all).
async fn post_fork(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    body: axum::body::Bytes,
) -> Result<Json<Value>, HttpError> {
    let upto: Option<String> = {
        let text = String::from_utf8_lossy(&body);
        if text.trim().is_empty() {
            None // NoContent arm
        } else {
            let v: Value = serde_json::from_str(&text).map_err(|_| tagged_bad_request())?;
            if !v.is_object() {
                return Err(tagged_bad_request()); // [] / "x" / 42 / null
            }
            match v.get("messageID") {
                None => None,
                Some(Value::String(s)) => Some(s.clone()),
                Some(_) => return Err(tagged_bad_request()), // non-string / null
            }
        }
    };
    let src_id = id.clone();
    let src = run_blocking(&st.db, move |db| {
        ocserve_store::load_session_wire(db, &src_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })?
    .ok_or_else(|| ApiError::not_found(format!("Session not found: {id}")))?;

    let new_id = ocserve_core::ids::ses_id();
    let now = now_ms();
    let worktree = st.paths["worktree"].as_str().unwrap_or("/").to_string();
    // createNext shape (session.ts): fresh id/slug, FORKED title, cost/tokens
    // zeroed (fork does not inherit usage), request-ctx directory/path —
    // workspaceID/metadata are not modeled by ocserve's schema (pre-existing).
    let info = json!({
        "id": new_id,
        "projectID": "global",
        "directory": worktree,
        "path": worktree.trim_start_matches('/'),
        "slug": slug_for(&new_id),
        "title": forked_title(src["title"].as_str().unwrap_or("")),
        "version": FREEZE_VERSION,
        "time": {"created": now, "updated": now},
        "cost": 0,
        "tokens": {"input": 0, "output": 0, "reasoning": 0,
                   "cache": {"read": 0, "write": 0}},
    });
    // PERF-10X F1: fork copies the whole source session (32k msgs measured)
    // — inline it parked a worker for the entire copy.
    let writer = st.writer.clone();
    let blobs = st.blobs.clone();
    let db = st.db.clone();
    let fork_info = info.clone();
    let fork_id = id.clone();
    let fork_upto = upto.clone();
    let stats = run_blocking(&db, move |db| {
        let env = ocserve_store::fork::ForkEnv {
            writer: &writer,
            blobs: Some(&*blobs),
            db,
        };
        ocserve_store::fork_session(
            &env,
            &fork_info,
            &fork_id,
            fork_upto.as_deref(),
            ocserve_core::ids::msg_id,
            ocserve_core::ids::prt_id,
        )
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })?;
    ocserve_metrics::counter("ocserve_session_fork_total", 1);
    tracing::info!(
        fork = %new_id,
        source = %id,
        upto = ?upto,
        messages = stats.messages,
        parts = stats.parts,
        "session forked"
    );
    // session.created AFTER the copy (D-FORK-1): upstream publishes at
    // createNext BEFORE cloning — a client reacting to the event can read a
    // partial session; ours fires when the history is complete. Every known
    // consumer (dialog-fork, ACP, oc-remote) REST-loads after this response.
    st.bus.publish(ocserve_core::event::frame(
        st.paths["directory"].as_str().unwrap_or("/"),
        "session.created",
        json!({"sessionID": new_id, "info": info}),
    ));
    Ok(Json(info))
}

/// GET /session/{id}/message — [{info, parts}] (captured wrapper shape).
async fn get_messages(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
    uri: axum::http::Uri,
    headers: axum::http::HeaderMap,
) -> Result<impl IntoResponse, HttpError> {
    // Freeze rule order (handlers/session.ts:110-117 + observed matrix in
    // message_page_contract.json): query-shape errors FIRST (Effect decodes
    // the whole query before the handler), then before-rules, then session.
    let limit: Option<u64> = match q.get("limit") {
        None => None,
        Some(raw) if raw.is_empty() => None, // `limit=` observed as full history
        Some(raw) => Some(parse_limit(raw).map_err(|message| HttpError::Query { message })?),
    };
    let before: Option<(String, i64)> = match (q.get("before").map(String::as_str), limit) {
        (Some(_), None) => return Err(tagged_bad_request()),
        (Some(b), Some(_)) => {
            Some(ocserve_store::decode_cursor(b).map_err(|_| tagged_bad_request())?)
        }
        (None, _) => None,
    };
    let exists_id = id.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })? {
        return Err(ApiError::not_found(format!("Session not found: {id}")).into());
    }
    // limit>0 → tuple-ordered cursor page (upstream MessageV2.page); absent/0
    // → full history. Headers ONLY when older messages remain (their rule).
    let mut page_n: Option<u64> = None;
    let mut page_next: Option<String> = None;
    let walk: ocserve_store::MessageWalk = match limit.filter(|n| *n > 0) {
        Some(n) => {
            let page_id = id.clone();
            let before_owned = before.clone();
            let (rows, _more, next) = run_blocking(&st.db, move |db| {
                ocserve_store::page_messages(
                    db,
                    &page_id,
                    n,
                    before_owned.as_ref().map(|(b, t)| (b.as_str(), *t)),
                )
            })
            .await
            .map_err(|e| ApiError {
                status: StatusCode::INTERNAL_SERVER_ERROR,
                name: "InternalError",
                message: format!("{e:#}"),
            })?;
            page_n = Some(n);
            page_next = next;
            ocserve_store::MessageWalk::Window(rows)
        }
        None => ocserve_store::MessageWalk::Seq { limit: None },
    };
    // F9: memoized paged body. Key carries db path (cross-test/process
    // isolation: same session id in another tmpdir must not collide) +
    // limit + cursor + the live write epoch (checked inside get).
    if page_n.is_some() && page_memo_enabled() {
        let memo_key = format!("{}\u{1}{id}\u{1}{limit:?}\u{1}{before:?}", st.db.display());
        let e0 = ocserve_store::write_epoch();
        {
            let mut m = page_memo().lock();
            if let Some(e) = m.get(&memo_key, e0) {
                drop(m);
                PAGE_MEMO_HITS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                return page_response(
                    &headers,
                    &uri,
                    &page_n,
                    &page_next,
                    axum::body::Body::from(e.body),
                )
                .map_err(Into::into);
            }
        }
        PAGE_MEMO_MISSES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        // assemble the page bounded — oversize aborts the collect and we
        // fall through to the original streaming path (walk not consumed:
        // the assembler gets a clone).
        let walk_asm = walk.clone();
        let db_asm = st.db.clone();
        let sid_asm = id.clone();
        let asm: anyhow::Result<std::result::Result<Vec<u8>, ()>> =
            run_blocking(&db_asm, move |db| {
                let mut buf: Vec<u8> = Vec::new();
                let mut total = 0usize;
                let mut first = true;
                let r = ocserve_store::for_each_message_json(db, &sid_asm, walk_asm, |chunk| {
                    let framed = if first {
                        first = false;
                        format!("[{chunk}")
                    } else {
                        format!(",{chunk}")
                    };
                    if total + framed.len() > PAGE_MEMO_ENTRY_CAP {
                        return Err(anyhow::anyhow!("page memo entry cap"));
                    }
                    total += framed.len();
                    buf.extend_from_slice(framed.as_bytes());
                    Ok(())
                });
                match r {
                    Ok(()) => {
                        // mirror the streaming close exactly: zero chunks
                        // => "[]" (a bare "]" would not be valid JSON)
                        if first {
                            buf.extend_from_slice(b"[]");
                        } else {
                            buf.push(b']');
                        }
                        Ok(Ok(buf))
                    }
                    Err(e) if e.to_string().contains("entry cap") => Ok(Err(())),
                    Err(e) => Err(e),
                }
            })
            .await;
        if let Ok(Ok(buf)) = asm {
            let bytes = bytes::Bytes::from(buf);
            let entry = PageEntry {
                body: bytes.clone(),
                next: page_next.clone(),
            };
            page_memo().lock().put(memo_key, entry, e0);
            return page_response(
                &headers,
                &uri,
                &page_n,
                &page_next,
                axum::body::Body::from(bytes),
            )
            .map_err(Into::into);
        }
        // oversize or assemble error → original streaming path below
    }
    // STREAMED response: one message per chunk through a bounded channel —
    // materializing a 16k-message session as Values OOM-killed the cgroup
    // (93MB response ≈ 400MB+ parsed; AGENTS §2.3 violation caught live).
    // Peak = channel(4) chunks + the message being built.
    let (tx, rx) = tokio::sync::mpsc::channel::<String>(4);
    let db = st.db.clone();
    let sid = id.clone();
    // F4: the tokio blocking pool, not a per-request OS thread — every
    // message_page request used to spawn a fresh thread (spawn+1MB VA per
    // page under load). The producer only does sync store work + channel
    // sends; dropping the JoinHandle keeps it detached exactly like before,
    // and the pool caps concurrent producers (bounded-by-construction).
    tokio::task::spawn_blocking(move || {
        let mut first = true;
        let r = ocserve_store::for_each_message_json(&db, &sid, walk, |chunk| {
            let framed = if first {
                first = false;
                format!("[{chunk}")
            } else {
                format!(",{chunk}")
            };
            tx.blocking_send(framed)
                .map_err(|_| anyhow::anyhow!("client disconnected"))
        });
        if let Err(e) = r {
            tracing::error!("message stream aborted session={sid}: {e:#}");
        }
        // best-effort close bracket (valid JSON even after a mid-stream error)
        let _ = tx.blocking_send(if first {
            "[]".to_string()
        } else {
            "]".to_string()
        });
    });
    let body_stream = futures_util::stream::unfold(rx, |mut rx| async move {
        rx.recv().await.map(|chunk| {
            (
                Ok::<axum::body::Bytes, std::convert::Infallible>(axum::body::Bytes::from(chunk)),
                rx,
            )
        })
    });
    page_response(
        &headers,
        &uri,
        &page_n,
        &page_next,
        axum::body::Body::from_stream(body_stream),
    )
    .map_err(Into::into)
}

/// PATCH /config + PATCH /global/config — W4 merge (v1 ConfigHttpApi.update:
/// deep-merges the payload, marks the instance for disposal, returns the
/// ECHOED payload — handlers/config.ts:18-21). Atomic file write (tmp in
/// the same dir → rename, .bak kept), then the boot-injected reloader
/// swaps derived payloads so the next GET serves fresh config. Divergence:
/// ocserve treats global==user config (single file); runtime endpoint
/// registries refresh on restart (TESTING §1.6).
async fn patch_config(
    State(st): State<Arc<AppState>>,
    Json(payload): Json<Value>,
) -> Result<Json<Value>, ApiError> {
    let path = config_write_path();
    let mut base: Value = if path.exists() {
        let raw = std::fs::read_to_string(&path).map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("read {}: {e}", path.display()),
        })?;
        serde_json::from_str(&raw).map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("parse {}: {e}", path.display()),
        })?
    } else {
        json!({})
    };
    deep_merge(&mut base, payload.clone());
    let dir = path.parent().ok_or_else(|| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: "config path has no parent".into(),
    })?;
    std::fs::create_dir_all(dir).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("mkdir {}: {e}", dir.display()),
    })?;
    let tmp = dir.join(format!(".opencode.json.tmp-{}", std::process::id()));
    let data = serde_json::to_vec_pretty(&base).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("serialize: {e}"),
    })?;
    std::fs::write(&tmp, &data).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("write {}: {e}", tmp.display()),
    })?;
    if path.exists() {
        let _ = std::fs::copy(&path, dir.join(".opencode.json.bak"));
    }
    std::fs::rename(&tmp, &path).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("rename {}: {e}", path.display()),
    })?;
    // instance disposal analog: rebuild derived payloads + registry AND
    // reconcile MCP (a PATCH that adds a server now connects it — H1
    // unifies every reload path through watch::reconcile)
    if let Err(e) = watch::reconcile(&st).await {
        tracing::warn!("config reload failed (restart to apply): {e:#}");
    }
    tracing::info!("config patched: {}", path.display());
    Ok(Json(payload))
}

/// POST /session/{id}/summarize — W3 compaction-lite (v1 handlers/session.ts
/// summarize: payload {providerID, modelID, auto?} → true). Runs ONE
/// transient text-only turn carrying buildPrompt's instruction + serialized
/// history (ocserve-core::compact); the summary lands as an ordinary
/// assistant message (divergence: no compaction-state/history filtering —
/// TESTING §1.6). Synchronous like upstream (`yield* promptSvc.loop`).
async fn post_summarize(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(sid): axum::extract::Path<String>,
    Json(payload): Json<Value>,
) -> Result<Json<Value>, ApiError> {
    let exists_id = sid.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })? {
        return Err(ApiError::not_found(format!("Session not found: {sid}")));
    }
    let provider = payload
        .get("providerID")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty());
    let model_id = payload
        .get("modelID")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty());
    let (Some(provider), Some(model_id)) = (provider, model_id) else {
        return Err(ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: "providerID and modelID are required".into(),
        });
    };

    let release = lock_session(&st, &sid).await?;
    let Ok(_guard) = release.arc().try_lock() else {
        return Err(session_busy(&sid));
    };

    // C5 retrofit (M6, COMPACTION.md §3): manual summarize = v1 source
    // (handlers/session.ts:273-293 at the FREEZE TAG v1.18.31, verified):
    // compactSvc.create(anchor with compaction part) + loop → pending task
    // → engine process → summary-exit. The run's outer machinery detects
    // the pending anchor — no generation runs (exit via last-message
    // summary check). Session-shaped history load ONLY for findLast user's
    // agent (v1 line 280).
    let history = run_blocking(&st.db, {
        let sid = sid.clone();
        move |db| ocserve_store::load_messages(db, &sid, None) // allow:load_messages (summarize compaction)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })?;
    let agent = history
        .iter()
        .rev()
        .find(|(info, _)| info["role"] == "user")
        .and_then(|(info, _)| info["agent"].as_str())
        .unwrap_or("build")
        .to_string();
    let prompt_payload = ocserve_core::wire::PromptRequest {
        model: Some(ocserve_core::wire::ModelRef {
            provider_id: provider.to_string(),
            model_id: model_id.to_string(),
        }),
        agent: Some(agent.clone()),
        ..Default::default()
    };
    let ctx = build_prompt_context(&st, &prompt_payload, &sid)?;
    let writer = st.writer.clone();
    ocserve_core::compaction::persist_anchor(&ctx, &writer, &sid, &agent, false, false)
        .await
        .map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("{e:#}"),
        })?;
    ocserve_core::prompt::run_prompt_with(
        &ctx,
        &writer,
        &sid,
        &prompt_payload,
        ocserve_core::prompt::RunOpts {
            // the ANCHOR is already the newest message — nothing from the
            // payload persists; pending→engine→summary-exit short-circuits
            // before any generation (persist_user=false keeps chat.message
            // silent, P0c invariant)
            persist_user: false,
            skip_history: false,
            prelude: None,
            tools_enabled: false,
            ..Default::default()
        },
    )
    .await
    .map_err(prompt_err)?;
    // P0e: session.compacted (magic-context consumes it after a summary)
    st.bus.publish(ocserve_core::event::frame(
        st.paths["directory"].as_str().unwrap_or("/"),
        "session.compacted",
        json!({"sessionID": sid}),
    ));
    Ok(Json(json!(true)))
}

/// Shared compaction trigger used by both the v1 summarize route and the v2
/// `/api/session/{id}/compact` route (single owner of the summarize flow).
pub(crate) async fn run_compact(
    st: &Arc<AppState>,
    ctx: &ocserve_core::prompt::PromptContext,
    writer: &ocserve_store::Writer,
    sid: &str,
) -> anyhow::Result<()> {
    let agent = ctx.agent.clone();
    let prompt_payload = ocserve_core::wire::PromptRequest {
        model: Some(ocserve_core::wire::ModelRef {
            provider_id: ctx.provider_id.clone(),
            model_id: ctx.model_id.clone(),
        }),
        agent: Some(agent.clone()),
        ..Default::default()
    };
    ocserve_core::compaction::persist_anchor(ctx, writer, sid, &agent, false, false).await?;
    ocserve_core::prompt::run_prompt_with(
        ctx,
        writer,
        sid,
        &prompt_payload,
        ocserve_core::prompt::RunOpts {
            persist_user: false,
            skip_history: false,
            prelude: None,
            tools_enabled: false,
            ..Default::default()
        },
    )
    .await?;
    st.bus.publish(ocserve_core::event::frame(
        st.paths["directory"].as_str().unwrap_or("/"),
        "session.compacted",
        json!({"sessionID": sid}),
    ));
    Ok(())
}

/// W5 admin bundle — POST /mcp/{name}/connect|disconnect (the client's
/// generic `/$action` path is identical to v1 McpPaths.connect/disconnect),
/// DELETE /mcp/{name}/auth, PUT/DELETE /auth/{providerID} (ocserve-OWN auth
/// overlay — NEVER the legacy auth.json ocserve reads), GET /provider/auth,
/// POST /global/dispose.
fn mcp_not_found(name: &str) -> HttpError {
    HttpError::TaggedData {
        status: StatusCode::NOT_FOUND,
        tag: "McpServerNotFoundError",
        fields: json!({"name": name, "message": format!("MCP server not found: {name}")}),
    }
}

async fn mcp_action(
    State(st): State<Arc<AppState>>,
    axum::extract::Path((name, action)): axum::extract::Path<(String, String)>,
) -> Result<Json<Value>, HttpError> {
    let hub = st.mcp.get().cloned().ok_or_else(|| HttpError::TaggedData {
        status: StatusCode::SERVICE_UNAVAILABLE,
        tag: "McpUnavailable",
        fields: json!({"message": "MCP hub not initialized"}),
    })?;
    let res = match action.as_str() {
        "connect" => hub.connect(&name).await,
        "disconnect" => hub.disconnect(&name).await,
        _ => Err(anyhow::anyhow!("unknown MCP action: {action}")),
    };
    res.map_err(|e| {
        if e.to_string().contains("not found") {
            mcp_not_found(&name)
        } else {
            HttpError::Api(ApiError {
                status: StatusCode::INTERNAL_SERVER_ERROR,
                name: "InternalError",
                message: format!("{e:#}"),
            })
        }
    })?;
    Ok(Json(json!(true)))
}

async fn mcp_auth_remove(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(name): axum::extract::Path<String>,
) -> Result<Json<Value>, HttpError> {
    // v1 authRemove: existence check first, then removeAuth → {success:true}
    let known = st
        .mcp
        .get()
        .map(|h| {
            let m = h.statuses();
            m.as_object()
                .map(|o| o.contains_key(&name))
                .unwrap_or(false)
                || h.known(&name)
        })
        .unwrap_or(false);
    if !known {
        return Err(mcp_not_found(&name));
    }
    // no OAuth MCP servers are configured → removing stored auth is a no-op
    // success (divergence TESTING §1.6)
    Ok(Json(json!({"success": true})))
}

/// ocserve-owned auth overlay — the file ocserve WRITES; the legacy auth.json
/// ocserve READS stays untouched (a bad write there would brick the TUI).
fn auth_overlay_path(st: &AppState) -> std::path::PathBuf {
    st.db
        .parent()
        .unwrap_or(std::path::Path::new("."))
        .join("auth-overlay.json")
}

fn write_json_atomic(path: &std::path::Path, value: &Value) -> Result<(), ApiError> {
    let dir = path.parent().ok_or_else(|| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: "path has no parent".into(),
    })?;
    std::fs::create_dir_all(dir).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("mkdir: {e}"),
    })?;
    let tmp = dir.join(format!(".tmp-{}-auth", std::process::id()));
    let data = serde_json::to_vec_pretty(value).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("serialize: {e}"),
    })?;
    std::fs::write(&tmp, data).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("write: {e}"),
    })?;
    std::fs::rename(&tmp, path).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("rename: {e}"),
    })?;
    Ok(())
}

async fn auth_put(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(pid): axum::extract::Path<String>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, ApiError> {
    if !body.is_object() {
        return Err(ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: "auth payload must be an object".into(),
        });
    }
    let path = auth_overlay_path(&st);
    let mut overlay: Value = if path.exists() {
        std::fs::read_to_string(&path)
            .ok()
            .and_then(|r| serde_json::from_str(&r).ok())
            .unwrap_or_else(|| json!({}))
    } else {
        json!({})
    };
    overlay[&pid] = body; // Auth.Info union stored verbatim (discriminator "type")
    write_json_atomic(&path, &overlay)?;
    if let Err(e) = watch::reconcile(&st).await {
        tracing::warn!("payload reload failed: {e:#}");
    }
    tracing::info!("auth overlay set provider={pid}");
    Ok(Json(json!(true)))
}

async fn auth_delete(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(pid): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
    let path = auth_overlay_path(&st);
    if path.exists() {
        let mut overlay: Value = std::fs::read_to_string(&path)
            .ok()
            .and_then(|r| serde_json::from_str(&r).ok())
            .unwrap_or_else(|| json!({}));
        if let Some(obj) = overlay.as_object_mut() {
            obj.remove(&pid);
        }
        write_json_atomic(&path, &overlay)?;
        if let Err(e) = watch::reconcile(&st).await {
            tracing::warn!("payload reload failed: {e:#}");
        }
    }
    tracing::info!("auth overlay remove provider={pid}");
    Ok(Json(json!(true)))
}

/// GET /provider/auth — {providerID: [{type,label}]}. v1 derives these from
/// auth HOOKS (provider/auth.ts:131); ocserve serves what ocserve can do —
/// api-key methods for every configured provider (divergence TESTING §1.6;
/// matches the PUT /auth {type:"api"} client flow).
async fn provider_auth_methods(State(st): State<Arc<AppState>>) -> Json<Value> {
    let payloads = st.payloads.read();
    let mut out = serde_json::Map::new();
    // LIVE shape: config_providers.providers is a LIST of {id, ...} — the
    // dict assumption returned {} (caught on the W5 live battery)
    if let Some(list) = payloads
        .config_providers
        .get("providers")
        .and_then(|p| p.as_array())
    {
        for e in list {
            if let Some(pid) = e.get("id").and_then(|v| v.as_str()) {
                out.insert(
                    pid.to_string(),
                    json!([{"type": "api", "label": "API Key"}]),
                );
            }
        }
    }
    Json(serde_json::Value::Object(out))
}

/// POST /global/dispose — v1 (global.ts:84): dispose + emit + true. Ocserve
/// analog: durable+live `server.instance.disposed` (EventReducer clears
/// client state) + payload reload. Divergence: no instance registry
/// (TESTING §1.6).
async fn global_dispose(State(st): State<Arc<AppState>>) -> Result<Json<Value>, ApiError> {
    // spec Event.server.instance.disposed.properties: {directory} (required).
    let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
    let props = json!({"directory": dir});
    ocserve_store::append_event(&st.writer, None, "server.instance.disposed", &props).map_err(
        |e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("{e:#}"),
        },
    )?;
    st.bus.publish(ocserve_core::event::frame(
        &dir,
        "server.instance.disposed",
        props,
    ));
    if let Err(e) = watch::reconcile(&st).await {
        tracing::warn!("payload reload failed: {e:#}");
    }
    tracing::info!("global dispose: emitted + payloads reloaded");
    Ok(Json(json!(true)))
}

/// POST /session/{id}/revert — real file-restoring revert (SNAPSHOT.md §2,
/// upstream session/revert.ts): snapshot the worktree if needed, restore the
/// pre-turn files, revert the tool `patch` parts, compute the revert-range
/// diff, persist per-message `summary.diffs` + session `summary`, emit
/// `session.diff`, set the marker. Busy → 409 (mapBusy parity).
async fn post_revert(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    Json(payload): Json<Value>,
) -> Result<axum::response::Response, ApiError> {
    let exists_id = id.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(internal_err)?
    {
        return Err(ApiError::not_found(format!("Session not found: {id}")));
    }
    let message_id = payload
        .get("messageID")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();
    if message_id.is_empty() {
        // freeze shape: {"name":"BadRequest","data":{"message":...,
        // "kind":"Payload"}} — probed 2026-10-08.
        return Ok((
            StatusCode::BAD_REQUEST,
            axum::Json(json!({
                "name": "BadRequest",
                "data": {"message": "Missing key\n  at [\"messageID\"]", "kind": "Payload"}
            })),
        )
            .into_response());
    }
    let part_id = payload
        .get("partID")
        .and_then(|v| v.as_str())
        .filter(|p| !p.is_empty())
        .map(String::from);

    // busy → 409 (upstream mapBusy)
    let release = lock_session(&st, &id).await?;
    let Ok(_guard) = release.arc().try_lock() else {
        return Err(session_busy(&id));
    };

    let marker = revert_apply(&st, &id, &message_id, part_id.as_deref()).await?;
    let _ = marker;
    let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();
    let _ = dir;
    let sid = id.clone();
    run_blocking(&st.db, move |db| ocserve_store::load_session_wire(db, &sid))
        .await
        .map_err(internal_err)?
        .map(|j| axum::Json(j).into_response())
        .ok_or_else(|| ApiError::not_found(format!("Session not found: {id}")))
}

pub(crate) fn internal_err(e: anyhow::Error) -> ApiError {
    ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    }
}

/// Shared revert engine (v1 + v2 stage): resolve target, restore files,
/// compute + persist the range summary, emit `session.diff`, set the marker.
pub(crate) async fn revert_apply(
    st: &Arc<AppState>,
    id: &str,
    message_id: &str,
    part_id: Option<&str>,
) -> Result<ocserve_core::revert::Marker, ApiError> {
    let db = st.db.clone();
    let sid = id.to_string();
    let mid = message_id.to_string();
    let pid = part_id.map(String::from);
    // Load the full message graph once, off the workers.
    let messages = run_blocking(&db, move |db| {
        ocserve_store::load_messages(db, &sid, None) // allow:load_messages (SNAPSHOT revert: full message graph for target resolution + patch collection, upstream revert.ts)
    })
    .await
    .map_err(internal_err)?;
    let target = ocserve_core::revert::resolve_target(&messages, &mid, pid.as_deref());
    let dir = session_directory_http(st, id).await;
    let snapshots = snapshots_for(st, &dir);

    // marker.snapshot = existing snapshot else capture now (upstream `?? track()`)
    let existing = run_blocking(&st.db, {
        let sid = id.to_string();
        move |db| ocserve_store::load_session_wire(db, &sid)
    })
    .await
    .map_err(internal_err)?
    .and_then(|s| {
        s.get("revert")
            .and_then(|r| r.get("snapshot"))
            .and_then(|v| v.as_str())
            .map(String::from)
    });

    let mut marker = ocserve_core::revert::Marker {
        message_id: target
            .as_ref()
            .map(|t| t.0.clone())
            .unwrap_or_else(|| mid.clone()),
        part_id: target.as_ref().and_then(|t| t.1.clone()),
        snapshot: existing.clone(),
        diff: None,
    };
    if let Some(snap) = snapshots.clone() {
        let prev = marker.snapshot.clone();
        let (snapshot, _patches) = run_blocking(&st.db, {
            let msgs = messages.clone();
            let snap = snap.clone();
            let prev = prev.clone();
            move |_db| {
                let snapshot = match &prev {
                    Some(s) => s.clone(),
                    None => snap.track().unwrap_or_default(),
                };
                if let Some(p) = &prev {
                    snap.restore(p);
                }
                let patches = ocserve_core::revert::collect_patches(&msgs);
                snap.revert(&patches);
                Ok::<_, anyhow::Error>((snapshot, patches))
            }
        })
        .await
        .map_err(internal_err)?;
        if !snapshot.is_empty() {
            marker.snapshot = Some(snapshot.clone());
            let diff = run_blocking(&st.db, {
                let snap = snap.clone();
                let s = snapshot.clone();
                move |_db| Ok::<_, anyhow::Error>(snap.diff(&s))
            })
            .await
            .map_err(internal_err)?;
            if !diff.is_empty() {
                marker.diff = Some(diff);
            }
        }
    }

    // revert-range diffs (summary.computeDiff over messages from the target)
    let range: Vec<(Value, Vec<Value>)> = match &target {
        Some((tmid, _)) => messages
            .iter()
            .skip_while(|(info, _)| info["id"].as_str() != Some(tmid.as_str()))
            .cloned()
            .collect(),
        None => Vec::new(),
    };
    let (diffs, summary) = match snapshots.clone() {
        Some(snap) => run_blocking(&st.db, move |_db| {
            Ok::<_, anyhow::Error>(ocserve_core::revert::compute_range(&snap, &range))
        })
        .await
        .map_err(internal_err)?,
        None => (
            Vec::new(),
            json!({"additions": 0, "deletions": 0, "files": 0}),
        ),
    };

    // persist: session summary + per-user-message summary.diffs
    let adds = summary["additions"].as_i64().unwrap_or(0);
    let dels = summary["deletions"].as_i64().unwrap_or(0);
    let files = summary["files"].as_i64().unwrap_or(0);
    ocserve_store::set_session_summary(&st.writer, id, adds, dels, files).map_err(internal_err)?;
    if let Some((tmid, _)) = &target
        && let Some((info, _)) = messages
            .iter()
            .find(|(i, _)| i["id"].as_str() == Some(tmid.as_str()))
    {
        let mut info = info.clone();
        info["summary"] = json!({"diffs": diffs});
        ocserve_store::update_message_info(&st.writer, id, tmid, &info).map_err(internal_err)?;
    }
    ocserve_store::set_session_revert(&st.writer, id, Some(&marker.to_value()))
        .map_err(internal_err)?;

    // events: session.diff (upstream Session.Event.Diff), then session.updated
    let diff_val = Value::Array(diffs.clone());
    emit_durable_http(
        st,
        id,
        "session.diff",
        json!({"sessionID": id, "diff": diff_val}),
    )?;
    let info = session_wire_or_partial_http(st, id, json!({"revert": marker.to_value()})).await;
    emit_durable_http(
        st,
        id,
        "session.updated",
        json!({"sessionID": id, "info": info}),
    )?;
    Ok(marker)
}

/// POST /session/{id}/unrevert — restore the snapshot + clear the marker.
async fn post_unrevert(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
    let exists_id = id.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(internal_err)?
    {
        return Err(ApiError::not_found(format!("Session not found: {id}")));
    }
    let release = lock_session(&st, &id).await?;
    let Ok(_guard) = release.arc().try_lock() else {
        return Err(session_busy(&id));
    };
    // restore the marker's snapshot (upstream unrevert)
    let existing = run_blocking(&st.db, {
        let sid = id.to_string();
        move |db| ocserve_store::load_session_wire(db, &sid)
    })
    .await
    .map_err(internal_err)?
    .and_then(|s| {
        s.get("revert")
            .and_then(|r| r.get("snapshot"))
            .and_then(|v| v.as_str())
            .map(String::from)
    });
    let dir = session_directory_http(&st, &id).await;
    if let Some(snap) = snapshots_for(&st, &dir)
        && let Some(s) = existing
    {
        let _ = run_blocking(&st.db, move |_db| {
            snap.restore(&s);
            Ok::<_, anyhow::Error>(())
        })
        .await;
    }
    ocserve_store::set_session_revert(&st.writer, &id, None).map_err(internal_err)?;
    let info = session_wire_or_partial_http(&st, &id, json!({"revert": Value::Null})).await;
    emit_durable_http(
        &st,
        &id,
        "session.updated",
        json!({"sessionID": id, "info": info}),
    )?;
    let sid = id.clone();
    run_blocking(&st.db, move |db| ocserve_store::load_session_wire(db, &sid))
        .await
        .map_err(internal_err)?
        .map(Json)
        .ok_or_else(|| ApiError::not_found(format!("Session not found: {id}")))
}

/// The session's stored working directory (or the server worktree fallback).
pub(crate) async fn session_directory_http(st: &Arc<AppState>, id: &str) -> String {
    let sid = id.to_string();
    run_blocking(&st.db, move |db| {
        Ok::<_, anyhow::Error>(ocserve_store::session_directory(db, &sid).ok().flatten())
    })
    .await
    .ok()
    .flatten()
    .unwrap_or_else(|| st.paths["directory"].as_str().unwrap_or("/").to_string())
}

/// POST /instance/dispose — freeze returns `true` (marks the instance for
/// disposal; a single-instance server's equivalent is publishing
/// server.instance.disposed like /global/dispose).
async fn instance_dispose(State(st): State<Arc<AppState>>) -> Result<Json<Value>, ApiError> {
    let _disposed = global_dispose(State(st)).await?;
    Ok(Json(json!(true)))
}

/// POST /session/{id}/init — runs the built-in `init` command
/// (command/index.ts Default.INIT, template initialize.txt) with the
/// caller's model. Freeze payload {messageID, providerID, modelID}; missing
/// model fields → 400 Effect Payload envelope ("Missing key at [modelID]").
async fn post_init(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    Json(payload): Json<Value>,
) -> Result<axum::response::Response, HttpError> {
    let provider = payload.get("providerID").and_then(|v| v.as_str());
    let model = payload.get("modelID").and_then(|v| v.as_str());
    let (Some(provider), Some(model)) = (provider, model) else {
        let missing = if provider.is_none() {
            "providerID"
        } else {
            "modelID"
        };
        return Ok((
            StatusCode::BAD_REQUEST,
            axum::Json(json!({
                "name": "BadRequest",
                "data": {
                    "message": format!("Missing key\n  at [\"{missing}\"]"),
                    "kind": "Payload"
                }
            })),
        )
            .into_response());
    };
    let exists_id = id.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })? {
        return Err(ApiError::not_found(format!("Session not found: {id}")).into());
    }
    // Reuse the /command path: command=init, arguments="", model from payload.
    let cmd_payload = json!({
        "command": "init",
        "arguments": "",
        "model": format!("{provider}/{model}"),
        "messageID": payload.get("messageID").cloned().unwrap_or(Value::Null),
    });
    post_command(State(st), axum::extract::Path(id), Json(cmd_payload))
        .await
        .map(|j| j.into_response())
}

/// POST /provider/{id}/oauth/authorize — ocserve has no OAuth flows; the
/// freeze shape for a non-OAuth provider is 400 {"name":"BadRequest",
/// "data":{}} (probed). OAuth login remains a named deferral (PLAN §17).
async fn post_provider_oauth(
    axum::extract::Path(_id): axum::extract::Path<String>,
) -> axum::response::Response {
    (
        StatusCode::BAD_REQUEST,
        axum::Json(json!({"name": "BadRequest", "data": {}})),
    )
        .into_response()
}

/// POST /provider/{id}/oauth/callback — same divergence as authorize.
async fn post_provider_oauth_callback(
    axum::extract::Path(_id): axum::extract::Path<String>,
) -> axum::response::Response {
    (
        StatusCode::BAD_REQUEST,
        axum::Json(json!({"name": "BadRequest", "data": {}})),
    )
        .into_response()
}

/// POST /mcp/{name}/auth/authenticate — unknown → 404 tagged; known (all
/// non-OAuth in practice) → 400 McpUnsupportedOAuthError (probe shapes).
async fn post_mcp_auth_authenticate(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(name): axum::extract::Path<String>,
) -> Result<axum::response::Response, HttpError> {
    mcp_auth_gate(&st, &name)?;
    Err(HttpError::TaggedData {
        status: StatusCode::BAD_REQUEST,
        tag: "McpUnsupportedOAuthError",
        fields: json!({"error": format!("MCP server {name} does not support OAuth")}),
    })
}

/// POST /mcp/{name}/auth/callback — same gate as authenticate.
async fn post_mcp_auth_callback(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(name): axum::extract::Path<String>,
) -> Result<axum::response::Response, HttpError> {
    mcp_auth_gate(&st, &name)?;
    Err(HttpError::TaggedData {
        status: StatusCode::BAD_REQUEST,
        tag: "McpUnsupportedOAuthError",
        fields: json!({"error": format!("MCP server {name} does not support OAuth")}),
    })
}

/// Shared existence check: unknown server → tagged 404 (probed shape).
fn mcp_auth_gate(st: &Arc<AppState>, name: &str) -> Result<(), HttpError> {
    let known = st
        .mcp
        .get()
        .map(|h| h.statuses().get(name).is_some())
        .unwrap_or(false);
    if known {
        return Ok(());
    }
    Err(HttpError::TaggedData {
        status: StatusCode::NOT_FOUND,
        tag: "McpServerNotFoundError",
        fields: json!({
            "name": name,
            "message": format!("MCP server not found: {name}"),
        }),
    })
}

/// GET /session/{id}/message/{mid} → {info, parts}; unknown → 404
/// NotFoundError envelope (probed freeze: {"name":"NotFoundError",
/// "data":{"message":"Message not found: <mid>"}}).
async fn get_message_by_id(
    State(st): State<Arc<AppState>>,
    axum::extract::Path((id, mid)): axum::extract::Path<(String, String)>,
) -> Result<axum::response::Response, ApiError> {
    let db = st.db.clone();
    let sid = id.clone();
    let mid_c = mid.clone();
    let found = run_blocking(&db, move |db| {
        ocserve_store::message_by_id(db, &sid, &mid_c)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })?;
    match found {
        Some(v) => Ok(axum::Json(v).into_response()),
        None => Err(ApiError::not_found(format!("Message not found: {mid}"))),
    }
}

/// POST /session/{id}/permissions/{pid} — deprecated v1 respond route; maps
/// onto the same gate as /permission/{id}/reply. Body {response: once|always|
/// reject}; unknown id → tagged PermissionNotFoundError (probed).
async fn post_session_permission_respond(
    State(st): State<Arc<AppState>>,
    axum::extract::Path((_id, pid)): axum::extract::Path<(String, String)>,
    Json(body): Json<Value>,
) -> Result<axum::response::Response, HttpError> {
    let reply = validate_permission_reply_key(&body, "response")?;
    if !st.gate.reply(&pid, reply) {
        return Err(HttpError::TaggedData {
            status: StatusCode::NOT_FOUND,
            tag: "PermissionNotFoundError",
            fields: json!({
                "requestID": pid,
                "message": format!("Permission request not found: {pid}"),
            }),
        });
    }
    Ok(axum::Json(json!(true)).into_response())
}

/// POST /log — v1 control-plane log writer; body {service,level,message,extra?}
/// → true (probed). Writes to the tracing target so it lands in the journal.
async fn post_log(Json(body): Json<Value>) -> Result<Json<Value>, HttpError> {
    let level = body.get("level").and_then(|v| v.as_str()).unwrap_or("info");
    let service = body
        .get("service")
        .and_then(|v| v.as_str())
        .unwrap_or("app");
    let message = body.get("message").and_then(|v| v.as_str()).unwrap_or("");
    match level {
        "debug" => tracing::debug!(target: "ocserve::client", service, "{message}"),
        "warn" => tracing::warn!(target: "ocserve::client", service, "{message}"),
        "error" => tracing::error!(target: "ocserve::client", service, "{message}"),
        _ => tracing::info!(target: "ocserve::client", service, "{message}"),
    }
    Ok(Json(json!(true)))
}

/// POST /session/search — W1 content search over part payloads (contract
/// block PLAN §17). ≥3 chars → trigram FTS substring MATCH; shorter → LIKE
/// fallback (same projection, one query either way). Inline AND blobbed
/// parts covered (projection stores uncompressed text).
async fn search_messages(
    State(st): State<Arc<AppState>>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, ApiError> {
    let query = body
        .get("query")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim()
        .to_string();
    if query.is_empty() || query.chars().count() > 512 {
        return Err(ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: "query must be a non-empty string of at most 512 characters".into(),
        });
    }
    let scope = body
        .get("sessionID")
        .and_then(|v| v.as_str())
        .map(str::to_string);
    let limit = body
        .get("limit")
        .and_then(|v| v.as_u64())
        .unwrap_or(50)
        .clamp(1, 200) as u32;
    let offset = body.get("offset").and_then(|v| v.as_u64()).unwrap_or(0) as u32;
    let needle = query.clone();
    let (hits, truncated) = run_blocking(&st.db, move |db| {
        ocserve_store::search_parts(db, &needle, scope.as_deref(), limit, offset)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })?;
    let out = json!({
        "hits": hits
            .iter()
            .map(|h| json!({
                "sessionID": h.session_id,
                "messageID": h.message_id,
                "partID": h.part_id,
                "role": h.role,
                "time": h.time,
                "snippet": ocserve_store::snippet(&h.text, &query),
            }))
            .collect::<Vec<_>>(),
        "truncated": truncated,
    });
    Ok(Json(out))
}

/// POST /session/{id}/message — sync prompt (blocks until stream completes,
/// returns {info, parts}; captured in testdata/m2/prompt_response.json).
/// v1 currentModel(sessionID): payload.model → session's stored model
/// (updated at prompt time) → configured default.
fn resolve_model(
    st: &Arc<AppState>,
    model: Option<&ocserve_core::wire::ModelRef>,
    session_id: &str,
) -> (String, String) {
    if let Some(m) = model {
        return (m.provider_id.clone(), m.model_id.clone());
    }
    if let Ok(conn) = ocserve_store::pragma::open_reader(&st.db) {
        let stored: Option<String> = conn
            .query_row(
                "SELECT model FROM session WHERE id = ?1",
                [session_id],
                |r| r.get(0),
            )
            .ok()
            .flatten();
        if let Some(raw) = stored
            && let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw)
        {
            let p = v.get("providerID").and_then(|x| x.as_str()).unwrap_or("");
            let m = v.get("id").and_then(|x| x.as_str()).unwrap_or("");
            if !p.is_empty() && !m.is_empty() {
                return (p.to_string(), m.to_string());
            }
        }
    }
    st.llm.read().default_model.clone()
}

/// Agent permission rules (v1 wire shape → evaluator Rule list). One owner so
/// the prompt loop and the v2 permission oracle cannot drift.
pub(crate) fn agent_rules(st: &Arc<AppState>, agent: &str) -> Vec<ocserve_tools::Rule> {
    st.payloads
        .read()
        .agent
        .iter()
        .find(|a| a["name"].as_str() == Some(agent))
        .and_then(|a| a["permission"].as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|r| {
                    Some(ocserve_tools::Rule {
                        permission: r["permission"].as_str()?.to_string(),
                        pattern: r["pattern"].as_str()?.to_string(),
                        action: r["action"].as_str()?.to_string(),
                    })
                })
                .collect()
        })
        .unwrap_or_default()
}

/// Skills for the system-prompt block (system.ts:skills): the command payloads
/// with source "skill", as (name, description, location). Built-in skills
/// carry an empty location; upstream writes "<built-in>" for those.
fn skill_entries(st: &Arc<AppState>) -> Vec<(String, String, String, String)> {
    let payloads = st.payloads.read();
    payloads
        .command
        .iter()
        .filter(|c| c.get("source").and_then(|v| v.as_str()) == Some("skill"))
        .filter_map(|c| {
            let name = c.get("name").and_then(|v| v.as_str())?.to_string();
            let description = c
                .get("description")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string();
            let location = c
                .get("location")
                .and_then(|v| v.as_str())
                .filter(|s| !s.is_empty())
                .unwrap_or("<built-in>")
                .to_string();
            let content = c
                .get("template")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string();
            Some((name, description, location, content))
        })
        .collect()
}

/// Resolve agent/model/system/endpoint/rules into a runnable prompt context.
/// Shared by POST /message (sync) and POST /prompt_async (backgrounded).
pub(crate) fn build_prompt_context(
    st: &Arc<AppState>,
    req: &ocserve_core::wire::PromptRequest,
    session_id: &str,
) -> Result<ocserve_core::prompt::PromptContext, ApiError> {
    let agent = req
        .agent
        .as_deref()
        .filter(|a| !a.is_empty())
        .map(String::from)
        .unwrap_or_else(|| st.llm.read().default_agent.clone());
    let system = st
        .llm
        .read()
        .systems
        .get(&agent)
        .cloned()
        .unwrap_or_default();
    let (pid, mid) = resolve_model(st, req.model.as_ref(), session_id);
    let (base_url, api_key) =
        st.llm
            .read()
            .endpoints
            .get(&pid)
            .cloned()
            .ok_or_else(|| ApiError {
                status: StatusCode::BAD_REQUEST,
                name: "BadRequest",
                message: format!("no endpoint configured for provider {pid}"),
            })?;
    let pricing = st
        .llm
        .read()
        .pricing
        .get(&(pid.clone(), mid.clone()))
        .copied();
    // agent permission rules (v1 wire shape → evaluator)
    let rules = agent_rules(st, &agent);
    let model_limit = st
        .llm
        .read()
        .limits
        .get(&(pid.clone(), mid.clone()))
        .cloned()
        .unwrap_or(serde_json::Value::Null);
    let compaction = {
        let payloads = st.payloads.read();
        ocserve_core::compact::CompactionCfg::from_value(Some(&payloads.compaction))
    };
    let compaction_system = st
        .llm
        .read()
        .systems
        .get("compaction")
        .cloned()
        .unwrap_or_else(|| {
            // summarize runs on the compaction agent; if it has no explicit
            // prompt, fall back to the model-selected provider prompt (what
            // upstream would build when agent.prompt is absent).
            ocserve_core::system_prompt::base_prompt(None, &mid, &pid)
        });
    let session_directory = ocserve_store::session_directory(&st.db, session_id)
        .ok()
        .flatten()
        .unwrap_or_else(|| st.paths["directory"].as_str().unwrap_or("/").to_string());
    // v1 parity (K-CAPDIFF): compose the provider system message exactly as
    // upstream — base (agent.prompt else model-selected provider prompt) +
    // environment block + skills block, "\n"-joined. Before this ocserve sent
    // a ~250-byte blurb where freeze sends ~9.5 KB (capdiff's first find).
    let base = ocserve_core::system_prompt::base_prompt(
        if system.is_empty() {
            None
        } else {
            Some(&system)
        },
        &mid,
        &pid,
    );
    let worktree = vcs::toplevel(std::path::Path::new(&session_directory))
        .unwrap_or_else(|| session_directory.clone());
    let is_git =
        worktree != session_directory || std::path::Path::new(&worktree).join(".git").exists();
    let env_block = ocserve_core::system_prompt::environment_block(
        &mid,
        &pid,
        &session_directory,
        &worktree,
        is_git,
        std::env::consts::OS,
        &js_to_date_string(now_ms()),
    );
    let skills = skill_entries(st);
    let skill_rows: Vec<(String, String, String)> = skills
        .iter()
        .map(|(n, d, l, _)| (n.clone(), d.clone(), l.clone()))
        .collect();
    let skills_blk = ocserve_core::system_prompt::skills_block(&skill_rows);
    let system = ocserve_core::system_prompt::compose(&base, &[Some(env_block), skills_blk], None);
    let snapshots = snapshots_for(st, &session_directory);
    let ctx = ocserve_core::prompt::PromptContext {
        db: st.db.clone(),
        blobs: st.blobs.clone(),
        bus: st.bus.clone(),
        // The session's OWN project directory (requested at create time), not
        // the server cwd — scopes tool ops + external_directory checks + event
        // routing to the client's project (the "session thinks it's in the
        // ocserve repo" + constant permission-prompt class).
        directory: session_directory,
        agent: agent.clone(),
        system,
        endpoint: ocserve_core::prompt::LlmEndpoint {
            base_url,
            api_key,
            pricing,
        },
        model_id: mid,
        provider_id: pid,
        rules,
        gate: st.gate.clone(),
        mcp: st.mcp.get().cloned(),
        plugins: st.plugins.get().cloned(),
        questions: st.question_gate.clone(),
        skills,
        model_limit,
        max_rounds: st.max_rounds,
        cost_ceiling: st.cost_ceiling,
        compaction,
        compaction_system,
        // SNAPSHOT (SNAPSHOT.md): config `snapshot != false` + env kill-switch;
        // the engine resolves the worktree lazily and no-ops when non-git.
        snapshots,
    };
    Ok(ctx)
}

/// Build the snapshot handle for a session directory. Enabled unless
/// `snapshot:false` in config (upstream key) or `OCSERVE_SNAPSHOT=0`. `None`
/// when disabled — the prompt then records no snapshots (upstream no-op parity).
fn snapshots_for(st: &AppState, directory: &str) -> Option<ocserve_core::snapshot::Snapshots> {
    let cfg_off = st
        .payloads
        .read()
        .config
        .get("snapshot")
        .map(|v| v == &serde_json::Value::Bool(false))
        .unwrap_or(false);
    let env_off = std::env::var("OCSERVE_SNAPSHOT")
        .map(|v| v == "0")
        .unwrap_or(false);
    if cfg_off || env_off {
        return None;
    }
    let data_dir = st.db.parent().unwrap_or_else(|| std::path::Path::new("."));
    Some(ocserve_core::snapshot::Snapshots::new(
        data_dir.to_path_buf(),
        directory,
        true,
    ))
}

/// Per-session prompt lock (insert-only while active; bounded by in-flight).
/// JS-Number-style rendering for Effect query error messages (JS prints
/// ≥1e21 in exponent form WITH a plus sign: "1e+21"; Rust's Display does
/// not — mirror observed upstream bytes).
fn fmt_js(v: f64) -> String {
    if v.is_finite() && v.abs() >= 1e21 {
        let raw = format!("{v:e}"); // e.g. 1e21 / -1.5e21
        if let Some(pos) = raw.find('e') {
            return format!("{}e+{}", &raw[..pos], &raw[pos + 1..]);
        }
        raw
    } else {
        format!("{v}")
    }
}

/// `?limit=` validation — freeze-probed matrix (message_page_contract.json):
/// absent/empty/0 → full history; integers (incl. "+5", "1e3") accepted up
/// to 2^53-1; negatives → range message; non-integers → integer message
/// (junk → "NaN"); mirror Effect's NumberFromString + Int + nonnegative
/// checks in observed order (integer-ness first, then ≥0). Rust's f64 parse
/// diverges from JS Number only on hex/whitespace forms — clients never
/// send those (documented, corpus note).
fn parse_limit(raw: &str) -> Result<u64, String> {
    const INT_MSG: &str = "Expected an integer, got {}\n  at [\"limit\"]";
    match raw.parse::<f64>() {
        Err(_) => Err(INT_MSG.replace("{}", "NaN")),
        Ok(v) if v.is_nan() => Err(INT_MSG.replace("{}", "NaN")),
        Ok(v) if v.is_infinite() => {
            Err(INT_MSG.replace("{}", if v > 0.0 { "Infinity" } else { "-Infinity" }))
        }
        Ok(v) if v.trunc() != v => Err(INT_MSG.replace("{}", &fmt_js(v))),
        Ok(v) if v.abs() > 9007199254740991.0 => Err(INT_MSG.replace("{}", &fmt_js(v))),
        Ok(v) if v < 0.0 => Err(format!(
            "Expected a value greater than or equal to 0, got {}\n  at [\"limit\"]",
            fmt_js(v)
        )),
        Ok(v) => Ok(v as u64),
    }
}

fn tagged_bad_request() -> HttpError {
    HttpError::Tagged {
        status: StatusCode::BAD_REQUEST,
        tag: "BadRequest",
    }
}

/// Abort-safe cleanup (antagonism A2): remove the prompt_locks entry when the
/// holder's task ends ANY way — normal completion, `?abort`, task panic,
/// delete_session — because post-run cleanup code never executes on abort
/// (entries only self-heal on the next prompt). Holds its own Arc clone so
/// the strong_count ≤ 2 check is exact during Drop. Concurrent holders or
/// waiters push the count above 2 → skip (entry stays — it's live).
struct LockRelease {
    locks: std::sync::Arc<
        parking_lot::Mutex<
            std::collections::HashMap<String, std::sync::Arc<tokio::sync::Mutex<()>>>,
        >,
    >,
    sid: String,
    lock: std::sync::Arc<tokio::sync::Mutex<()>>,
}

impl LockRelease {
    fn arc(&self) -> &std::sync::Arc<tokio::sync::Mutex<()>> {
        &self.lock
    }
}

impl Drop for LockRelease {
    fn drop(&mut self) {
        // EXACT reference accounting (the first design had callers holding a
        // separate Arc clone → drop-order dependent skip, caught by the abort
        // test): ALL locking goes through arc() so the only refs are the map
        // entry + this release. remove iff count ≤ 2 (map + self).
        let mut map = self.locks.lock();
        if let Some(arc) = map.get(&self.sid)
            && std::sync::Arc::strong_count(arc) <= 2
        {
            map.remove(&self.sid);
        }
    }
}

/// Session prompt lock + abort-safe cleanup in ONE object: callers lock via
/// `release.arc()` (never a second Arc clone) so Drop's count is exact, and
/// dropping release — normal end, `?`, panic, or task abort — evicts the map
/// entry. Concurrent holders/waiters each hold their own release (count > 2)
/// so live entries are never evicted.
pub(crate) async fn lock_session(st: &Arc<AppState>, sid: &str) -> Result<LockRelease, ApiError> {
    let lock = {
        let mut map = st.prompt_locks.lock();
        if map.len() >= 64 && !map.contains_key(sid) {
            return Err(ApiError {
                status: StatusCode::SERVICE_UNAVAILABLE,
                name: "ServiceUnavailableError",
                message: "too many concurrent prompts".into(),
            });
        }
        map.entry(sid.to_string())
            .or_insert_with(|| std::sync::Arc::new(tokio::sync::Mutex::new(())))
            .clone()
    };
    Ok(LockRelease {
        locks: std::sync::Arc::clone(&st.prompt_locks),
        sid: sid.to_string(),
        lock,
    })
}

pub(crate) fn prompt_err(e: anyhow::Error) -> ApiError {
    let msg = format!("{e:#}");
    if msg.starts_with("Session not found") {
        ApiError::not_found(msg)
    } else {
        ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: msg,
        }
    }
}

async fn post_message(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    Json(payload): Json<Value>,
) -> Result<impl IntoResponse, HttpError> {
    // wire decode FIRST (freeze: HttpApi payload decode precedes handler —
    // probe: `{}` on a nonexistent session → 400, valid body → 404).
    let req = decode_v1(ocserve_core::wire::PromptRequest::decode(&payload))?;
    let ctx = build_prompt_context(&st, &req, &id)?;
    let _release = lock_session(&st, &id).await?;
    let _guard = _release.arc().lock().await;
    let writer = st.writer.clone();
    let result = ocserve_core::prompt::run_prompt(&ctx, &writer, &id, &req)
        .await
        .map_err(prompt_err)?;
    drop(_guard);
    let (info, parts) = result;
    Ok(Json(json!({"info": info, "parts": parts})))
}

/// POST /session/{id}/prompt_async — v1 `session.prompt_async` (oc-remote's
/// send path): same PromptPayload as /message, **204 immediately**, prompt
/// runs in the background; results arrive on SSE. Errors: NotFound (bad
/// session) / BadRequest before spawn; background failures are logged
/// (upstream handlers/session.ts logs "prompt_async failed", no reply body).
async fn post_prompt_async(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    Json(payload): Json<Value>,
) -> Result<StatusCode, HttpError> {
    // decode FIRST (probe: `prompt_async {}` → 400 Missing key, never 204;
    // HttpApi payload decode precedes both 204 and the session check).
    let req = decode_v1(ocserve_core::wire::PromptRequest::decode(&payload))?;
    let exists_id = id.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })? {
        return Err(ApiError::not_found(format!("Session not found: {id}")).into());
    }
    let ctx = build_prompt_context(&st, &req, &id)?;
    // release MOVES INTO the task: the handler returns 204 immediately, so a
    // handler-scoped guard would drop (and evict the lock entry) while the
    // prompt is still running (antagonism A2 — ownership, not just RAII).
    let release = lock_session(&st, &id).await?;
    let writer = st.writer.clone();
    let sid = id.clone();
    let generation = st
        .prompt_gen
        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let tasks = st.prompt_tasks.clone();
    let tasks_insert = st.prompt_tasks.clone();
    let handle = tokio::spawn(async move {
        let _release = release;
        let guard = _release.arc().lock().await;
        match ocserve_core::prompt::run_prompt(&ctx, &writer, &sid, &req).await {
            Ok(_) => {}
            Err(e) => tracing::error!("prompt_async failed session={sid}: {e:#}"),
        }
        drop(guard);
        let mut t = tasks.lock();
        if let Some((g, _)) = t.get(&sid)
            && *g == generation
        {
            t.remove(&sid);
        }
    });
    tasks_insert.lock().insert(id.clone(), (generation, handle));
    Ok(StatusCode::NO_CONTENT)
}

// ---- oc-remote contract family (Batch 1: session/message/part) ----

async fn get_children(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
    let exists_id = id.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })? {
        return Err(ApiError::not_found(format!("Session not found: {id}")));
    }
    run_blocking(&st.db, move |db| ocserve_store::load_children(db, &id))
        .await
        .map(|v| Json(Value::Array(v)))
        .map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("{e:#}"),
        })
}

async fn get_todos(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
    // freeze 404s an unknown session (probe); ocserve returned `[]`.
    let exists_id = id.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })? {
        return Err(ApiError::not_found(format!("Session not found: {id}")));
    }
    run_blocking(&st.db, move |db| ocserve_store::load_todos(db, &id))
        .await
        .map(|v| Json(Value::Array(v)))
        .map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("{e:#}"),
        })
}

/// PATCH /session/{id} — oc-remote rename sends {title} only. Other
/// UpdatePayload fields (metadata/permission/time.archived) are not stored
/// by ocserve — accepted and ignored (documented divergence).
async fn patch_session(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    Json(payload): Json<Value>,
) -> Result<Json<Value>, ApiError> {
    let exists_id = id.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })? {
        return Err(ApiError::not_found(format!("Session not found: {id}")));
    }
    let mut dirty = false;
    if let Some(title) = payload.get("title").and_then(|t| t.as_str()) {
        ocserve_store::update_session_title(&st.writer, &id, title).map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("{e:#}"),
        })?;
        dirty = true;
    }
    // K-MODEL-STATE: freeze Patch semantics for model/agent (oc-remote only
    // calls title; TUI/other clients patch model+agent). permission is
    // deliberately NOT patchable (no caller in evidence — ocserve's internal
    // key format differs from upstream's ruleset; divergence noted K row).
    if payload.get("model").is_some() {
        let raw = payload.get("model").cloned().unwrap_or(Value::Null);
        let model_json: Value = match raw {
            Value::Null => Value::Null, // clear → SQL NULL → resolve falls to default
            Value::Object(ref m) => {
                let ok = m
                    .get("id")
                    .and_then(|v| v.as_str())
                    .is_some_and(|v| !v.is_empty())
                    && m.get("providerID")
                        .and_then(|v| v.as_str())
                        .is_some_and(|v| !v.is_empty());
                if !ok {
                    return Err(ApiError {
                        status: StatusCode::BAD_REQUEST,
                        name: "BadRequest",
                        message: "model must be {id, providerID, variant?} or null".into(),
                    });
                }
                Value::String(
                    serde_json::to_string(&json!({
                        "id": m["id"],
                        "providerID": m["providerID"],
                        "variant": m.get("variant").cloned().unwrap_or(json!("default")),
                    }))
                    .map_err(|e| ApiError {
                        status: StatusCode::INTERNAL_SERVER_ERROR,
                        name: "InternalError",
                        message: format!("{e:#}"),
                    })?,
                )
            }
            _ => {
                return Err(ApiError {
                    status: StatusCode::BAD_REQUEST,
                    name: "BadRequest",
                    message: "model must be {id, providerID, variant?} or null".into(),
                });
            }
        };
        st.writer
            .write(vec![ocserve_store::WriteOp::Sql {
                sql: "UPDATE session SET model = ?2 WHERE id = ?1".into(),
                params: vec![id.clone().into(), model_json],
            }])
            .map_err(|e| ApiError {
                status: StatusCode::INTERNAL_SERVER_ERROR,
                name: "InternalError",
                message: format!("{e:#}"),
            })?;
        dirty = true;
    }
    if payload.get("agent").is_some() {
        let agent = match payload.get("agent") {
            Some(Value::Null) => String::new(),
            Some(Value::String(a)) if !a.is_empty() => a.clone(),
            _ => {
                return Err(ApiError {
                    status: StatusCode::BAD_REQUEST,
                    name: "BadRequest",
                    message: "agent must be a non-empty string or null".into(),
                });
            }
        };
        st.writer
            .write(vec![ocserve_store::WriteOp::Sql {
                sql: "UPDATE session SET agent = ?2 WHERE id = ?1".into(),
                params: vec![id.clone().into(), agent.into()],
            }])
            .map_err(|e| ApiError {
                status: StatusCode::INTERNAL_SERVER_ERROR,
                name: "InternalError",
                message: format!("{e:#}"),
            })?;
        dirty = true;
    }
    let info_id = id.clone();
    let info = run_blocking(&st.db, move |db| {
        ocserve_store::load_session_wire(db, &info_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })?
    .ok_or_else(|| ApiError::not_found(format!("Session not found: {id}")))?;
    if dirty {
        // FULL session (already loaded above) so the event carries every
        // required key — a partial {id,title,model,agent} merges into client
        // stores and drops `time`/`version` (TUI/SDK store corruption).
        st.bus.publish(ocserve_core::event::frame(
            st.paths["directory"].as_str().unwrap_or("/"),
            "session.updated",
            json!({"sessionID": id, "info": info}),
        ));
    }
    Ok(Json(info))
}

/// DELETE /session/{id} → true (v1 Schema.Boolean). Aborts any in-flight
/// async prompt first (it would write into a deleted session otherwise).
async fn delete_session_route(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
    if let Some((_, handle)) = st.prompt_tasks.lock().remove(&id) {
        handle.abort();
    }
    let found = ocserve_store::delete_session(&st.writer, &st.db, &id).map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })?;
    if !found {
        return Err(ApiError::not_found(format!("Session not found: {id}")));
    }
    // RAM audit 2026-10-11: release per-session gate state (always-grants +
    // hydrated marker) — it otherwise accumulates for every session ever seen.
    st.gate.forget_session(&id);
    // P0e: magic-context consumes session.deleted (live publish — the
    // durable rows cascade away with the session).
    st.bus.publish(ocserve_core::event::frame(
        st.paths["directory"].as_str().unwrap_or("/"),
        "session.deleted",
        json!({"sessionID": id}),
    ));
    Ok(Json(json!(true)))
}

/// POST /session/{id}/abort → true (idempotent: no active task = still true).
async fn post_abort(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, ApiError> {
    if let Some((_, handle)) = st.prompt_tasks.lock().remove(&id) {
        handle.abort();
        tracing::info!("abort: cancelled prompt task for {id}");
    }
    Ok(Json(json!(true)))
}

async fn delete_message_route(
    State(st): State<Arc<AppState>>,
    axum::extract::Path((sid, mid)): axum::extract::Path<(String, String)>,
) -> Result<Json<Value>, ApiError> {
    let found =
        ocserve_store::delete_message(&st.writer, &st.db, &sid, &mid).map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("{e:#}"),
        })?;
    if !found {
        return Err(ApiError::not_found(format!("Message not found: {mid}")));
    }
    // P0e: magic-context consumes message.removed (live publish — durable
    // rows cascade away with the message).
    st.bus.publish(ocserve_core::event::frame(
        st.paths["directory"].as_str().unwrap_or("/"),
        "message.removed",
        json!({"sessionID": sid, "messageID": mid}),
    ));
    Ok(Json(json!(true)))
}

async fn delete_part_route(
    State(st): State<Arc<AppState>>,
    axum::extract::Path((_sid, mid, pid)): axum::extract::Path<(String, String, String)>,
) -> Result<Json<Value>, ApiError> {
    ocserve_store::delete_part(&st.writer, &st.db, &mid, &pid)
        .map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("{e:#}"),
        })?
        .then_some(Json(json!(true)))
        .ok_or_else(|| ApiError::not_found(format!("Part not found: {pid}")))
}

/// PATCH .../part/{pid} — replace part JSON (native sessionID/messageID keys;
/// v1 verifies all three ids match the path — oc-remote pre-validates too).
async fn patch_part_route(
    State(st): State<Arc<AppState>>,
    axum::extract::Path((sid, mid, pid)): axum::extract::Path<(String, String, String)>,
    Json(part): Json<Value>,
) -> Result<Json<Value>, ApiError> {
    let body_sid = part.get("sessionID").and_then(|v| v.as_str()).unwrap_or("");
    let body_mid = part.get("messageID").and_then(|v| v.as_str()).unwrap_or("");
    let body_pid = part.get("id").and_then(|v| v.as_str()).unwrap_or("");
    if body_sid != sid || body_mid != mid || body_pid != pid {
        return Err(ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: "part id/sessionID/messageID must match the path".into(),
        });
    }
    ocserve_store::update_part(&st.writer, &st.blobs, &sid, &mid, &pid, &part)
        .map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("{e:#}"),
        })?
        .then_some(Json(part))
        .ok_or_else(|| ApiError::not_found(format!("Part not found: {pid}")))
}

// ---- oc-remote contract family (Batch 2: file/find/question) ----

fn scope_path(base: &std::path::Path, raw: &str) -> Result<std::path::PathBuf, ApiError> {
    let joined = if raw.is_empty() || raw == "." {
        base.to_path_buf()
    } else {
        let p = std::path::Path::new(raw);
        if p.is_absolute() {
            p.to_path_buf()
        } else {
            base.join(p)
        }
    };
    let canon_base = std::fs::canonicalize(base).unwrap_or_else(|_| base.to_path_buf());
    // Lexical containment FIRST: normalize `..` without touching the FS so an
    // escape is rejected deterministically even when the target is missing
    // (the previous code fell back to the raw `..`-laden path when canonicalize
    // failed, and a lexically-`starts_with` check on that path passed for a
    // traversal like `base/../../../etc`). Both the normalized join AND the
    // canonicalized real path must stay inside the base.
    let norm = lexically_normalize(&joined);
    if !norm.starts_with(&canon_base) && !norm.starts_with(base) {
        return Err(ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: format!("path escapes the session directory: {raw}"),
        });
    }
    let canon = std::fs::canonicalize(&joined).unwrap_or(norm);
    if !canon.starts_with(&canon_base) {
        return Err(ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: format!("path escapes the session directory: {raw}"),
        });
    }
    Ok(canon)
}

/// Lexically resolve `.`/`..` (no filesystem access).
fn lexically_normalize(p: &std::path::Path) -> std::path::PathBuf {
    use std::path::Component;
    let mut out = std::path::PathBuf::new();
    for c in p.components() {
        match c {
            Component::ParentDir => {
                out.pop();
            }
            Component::CurDir => {}
            other => out.push(other.as_os_str()),
        }
    }
    out
}

/// GET /file?path= — directory listing (FileNode list, oc-remote file browser).
async fn list_directory(
    State(st): State<Arc<AppState>>,
    axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Value>, ApiError> {
    let base = std::path::Path::new(st.paths["directory"].as_str().unwrap_or("/"));
    let raw = q.get("path").map(String::as_str).unwrap_or(".");
    let dir = scope_path(base, raw)?;
    // M3 (same class as find_files): `read_dir` + per-entry `metadata()` is
    // blocking syscalls, inline on a tokio worker. A directory with thousands
    // of entries would park the worker for the whole loop.
    let dir_for_worker = dir.clone();
    let raw_owned = raw.to_string();
    let rd = tokio::task::spawn_blocking(move || std::fs::read_dir(&dir_for_worker))
        .await
        .map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalServerError",
            message: format!("list worker: {e}"),
        })?
        .map_err(|e| ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: format!("list {}: {e}", dir.display()),
        })?;
    let mut nodes: Vec<Value> = Vec::new();
    for entry in rd.flatten() {
        let name = entry.file_name().to_string_lossy().to_string();
        let path = entry.path();
        let meta = entry.metadata().ok();
        let is_dir = meta.as_ref().map(|m| m.is_dir()).unwrap_or(false);
        let rel = if raw_owned.is_empty() || raw_owned == "." {
            name.clone()
        } else {
            format!("{}/{}", raw_owned.trim_end_matches('/'), name)
        };
        let ignored =
            name.starts_with('.') || name == "node_modules" || name == "target" || name == ".git";
        // Freeze FileNode = {name,path,absolute,type,ignored} with
        // `additionalProperties:false` (spec probed 2026-10-09). ocserve
        // previously added `size`/`modified`, which freeze never emits and
        // no client can depend on — the spec-validation oracle (P4) caught
        // it. Emit exactly the freeze keys.
        nodes.push(json!({
            "name": name,
            "path": rel,
            "type": if is_dir { "directory" } else { "file" },
            "absolute": path.to_string_lossy(),
            "ignored": ignored,
        }));
    }
    // directories first, then name (browser expectations), bounded
    nodes.sort_by(|a, b| {
        let da = a["type"] == "directory";
        let db = b["type"] == "directory";
        db.cmp(&da)
            .then_with(|| a["name"].as_str().cmp(&b["name"].as_str()))
    });
    nodes.truncate(5000);
    Ok(Json(Value::Array(nodes)))
}

/// GET /file/content?path= — {type: text|binary, content, encoding}.
async fn read_file_content(
    State(st): State<Arc<AppState>>,
    axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Value>, HttpError> {
    let base = std::path::Path::new(st.paths["directory"].as_str().unwrap_or("/"));
    let raw = q
        .get("path")
        .map(String::as_str)
        .ok_or_else(|| HttpError::Query {
            message: "Missing key\n  at [\"path\"]".into(),
        })?;
    let path = scope_path(base, raw).map_err(HttpError::Api)?;
    // freeze returns an empty text body for a missing file (probed), not an
    // error — the web UI probes optional paths.
    // Perf dive 2026-10-11: the read ran INLINE on a tokio worker (up to 1 MB
    // blocking fs + copy — the M3 class). Move it to the blocking pool.
    let read_path = path.clone();
    let bytes = match tokio::task::spawn_blocking(move || std::fs::read(&read_path)).await {
        Ok(Ok(b)) => b,
        _ => {
            return Ok(Json(json!({"type": "text", "content": ""})));
        }
    };
    // bound: MEMORY §6 parse cap — truncate at 1MB, then floor to a UTF-8 char
    // boundary so a multibyte char straddling the cutoff does not make a text
    // file mislabel as "binary".
    let truncated = bytes.len() > 1024 * 1024;
    let bytes = if truncated {
        let mut cut = 1024 * 1024;
        while cut > 0 && (bytes[cut] & 0xC0) == 0x80 {
            cut -= 1;
        }
        &bytes[..cut]
    } else {
        &bytes[..]
    };
    let (kind, content) = match std::str::from_utf8(bytes) {
        Ok(text) => ("text", text.to_string()),
        Err(_) => ("binary", String::from_utf8_lossy(bytes).into_owned()),
    };
    Ok(Json(json!({
        "type": kind,
        "content": content,
        "encoding": "utf-8",
    })))
}

/// GET /find/file?query=&type=&limit= — file-name/path search → List<String>
/// (bounded walk; case-insensitive substring, glob when the query has *).
async fn find_files(
    State(st): State<Arc<AppState>>,
    axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Value>, HttpError> {
    let base = std::path::Path::new(st.paths["directory"].as_str().unwrap_or("/")).to_path_buf();
    // `query` is REQUIRED (freeze Query decode: missing → 400, kind Query);
    // an empty value is accepted (→ no matches).
    let query = q
        .get("query")
        .ok_or_else(|| HttpError::Query {
            message: "Missing key\n  at [\"query\"]".into(),
        })?
        .to_lowercase();
    if query.is_empty() {
        return Ok(Json(json!([])));
    }
    let want_type = q.get("type").map(String::as_str).unwrap_or("").to_string();
    // freeze default limit = 10 (ripgrep find limit, file.ts:47)
    let limit: usize = q
        .get("limit")
        .and_then(|l| l.parse().ok())
        .unwrap_or(10)
        .clamp(1, 200);
    let glob_mode = query.contains('*') || query.contains('?');
    // M3: the walk is a blocking filesystem scan (bounded at 20,000 dirs, but
    // a big worktree still means thousands of `read_dir` syscalls per
    // request). It ran INLINE on a tokio worker — the same convoy class as the
    // pre-F1 SQLite work, which is exactly what `run_blocking` exists for.
    // The load harness never caught this because its LIST_PATH is an empty
    // /tmp dir; on a real worktree it would park a worker.
    let out = tokio::task::spawn_blocking(move || {
        walk_files(&base, &query, &want_type, limit, glob_mode)
    })
    .await
    .map_err(|e| {
        HttpError::Api(ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalServerError",
            message: format!("find_files worker: {e}"),
        })
    })?
    .map_err(|e| {
        HttpError::Api(ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: format!("find_files: {e:#}"),
        })
    })?;
    Ok(Json(Value::Array(
        out.into_iter().map(Value::String).collect(),
    )))
}

/// Thread the last `walk_files` executed on (M3 test seam).
///
/// Global, not `thread_local!`: the walk runs on a *blocking-pool* thread, so
/// a thread-local record would be written on one thread and read on another —
/// the first version of this seam was invisible for exactly that reason (caught
/// by the test asserting on it).
static WALK_THREAD: std::sync::Mutex<Option<std::thread::ThreadId>> = std::sync::Mutex::new(None);

/// Blocking body of `GET /find/file` — a bounded, single-threaded directory
/// walk. Existing bounds kept exactly (20,000 dirs visited, stop once the
/// result set reaches 4×limit); only the *thread* it runs on changed.
pub(crate) fn walk_files(
    base: &std::path::Path,
    query: &str,
    want_type: &str,
    limit: usize,
    glob_mode: bool,
) -> anyhow::Result<Vec<String>> {
    // M3 test seam: which thread executed the walk. Async tokio workers and
    // the blocking pool are different threads, so comparing this with the
    // thread a request body runs on proves the walk was moved OFF the worker
    // without depending on wall-clock timing (a small tree walks in
    // microseconds — the first version of this test used a ticker and passed
    // even with the walk inline, i.e. it was not discriminating).
    if let Ok(mut g) = WALK_THREAD.lock() {
        *g = Some(std::thread::current().id());
    }
    let mut out: Vec<String> = Vec::new();
    let mut stack = vec![(base.to_path_buf(), String::new())];
    let mut visited = 0usize;
    while let Some((dir, rel)) = stack.pop() {
        visited += 1;
        if visited > 20_000 || out.len() >= limit * 4 {
            break; // bounded (AGENTS §2.3)
        }
        let Ok(rd) = std::fs::read_dir(&dir) else {
            continue;
        };
        for entry in rd.flatten() {
            let name = entry.file_name().to_string_lossy().to_string();
            if name == ".git" || name == "node_modules" {
                continue;
            }
            let is_dir = entry.file_type().map(|t| t.is_dir()).unwrap_or(false);
            let child_rel = if rel.is_empty() {
                name.clone()
            } else {
                format!("{rel}/{name}")
            };
            let matched = if glob_mode {
                ocserve_tools::wildcard_match(&name, query)
            } else {
                child_rel.to_lowercase().contains(query)
            };
            if matched
                && (want_type.is_empty()
                    || (want_type == "dir" && is_dir)
                    || (want_type == "file" && !is_dir))
            {
                out.push(child_rel.clone());
            }
            if is_dir {
                stack.push((entry.path(), child_rel));
            }
        }
    }
    // prefix matches first, then lexicographic; capped
    out.sort_by(|a, b| {
        let pa = a.to_lowercase().starts_with(query);
        let pb = b.to_lowercase().starts_with(query);
        pb.cmp(&pa).then_with(|| a.cmp(b))
    });
    out.truncate(limit);
    Ok(out)
}

/// GET /find?pattern= — text search → SearchMatch list. Single `grep -rnE`
/// spawn (bounded output, 5s); absoluteOffset is not derivable from grep
/// output and this route has no callers in oc-remote — recorded as 0
/// (documented, not silently invented).
async fn find_text(
    State(st): State<Arc<AppState>>,
    axum::extract::Query(q): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Value>, HttpError> {
    // `pattern` is REQUIRED (freeze Query decode: missing → 400, kind Query);
    // empty value accepted (→ no matches).
    let pattern = q.get("pattern").ok_or_else(|| HttpError::Query {
        message: "Missing key\n  at [\"pattern\"]".into(),
    })?;
    if pattern.is_empty() {
        return Ok(Json(json!([])));
    }
    let base = std::path::Path::new(st.paths["directory"].as_str().unwrap_or("/")).to_path_buf();
    let pattern = pattern.to_string();
    // Freeze uses ripgrep (gitignore-aware). ocserve shells to grep, so we must
    // approximate the ignore behaviour: skip VCS/build/vendor dirs. Without
    // this, `/find` scanned `target/` (gigabytes) and hung the request — a real
    // bug (freeze returns matches instantly).
    const EXCLUDES: &[&str] = &[
        ".git",
        "node_modules",
        "target",
        ".venv",
        "venv",
        "__pycache__",
        "dist",
        "build",
        ".terraform",
        ".terragrunt-cache",
    ];
    let base2 = base.clone();
    // Two bounded grep passes (both on the blocking pool):
    //   A: line-level → path:line:lineByte:content  (path.text/lines.text/line_number/absolute_offset)
    //   B: match-level → path:line:matchByte:matchText  (submatches[{match.text,start,end}])
    // freeze's SearchMatch is {path:{text}, lines:{text}, line_number,
    // absolute_offset, submatches:[{match:{text}, start, end}]} with start/end
    // as byte offsets WITHIN the line (ripgrep.ts:56-72).
    let out = tokio::task::spawn_blocking(move || {
        let run = |extra: &[&str]| -> std::io::Result<std::process::Output> {
            let mut c = std::process::Command::new("grep");
            c.args(["-rnE", "--binary-files=without-match", "-m", "50"]);
            c.args(extra);
            for d in EXCLUDES {
                c.arg(format!("--exclude-dir={d}"));
            }
            c.arg("--").arg(&pattern).arg(&base2);
            c.env("LC_ALL", "C").output()
        };
        // A first (line-level, -n) then B (-o -b for each match offset)
        let a = run(&["-n"])?;
        let b = run(&["-on", "-b"])?;
        Ok::<_, std::io::Error>((a, b))
    })
    .await
    .map_err(|e| {
        HttpError::Api(ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("find worker: {e}"),
        })
    })?;
    let (a_out, b_out) = match out {
        Ok(o) => o,
        Err(e) => {
            return Err(HttpError::Api(ApiError {
                status: StatusCode::INTERNAL_SERVER_ERROR,
                name: "InternalError",
                message: format!("grep spawn: {e}"),
            }));
        }
    };
    if a_out.status.code().unwrap_or(0) > 1 {
        return Err(HttpError::Api(ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: "invalid pattern".into(),
        }));
    }
    // match-level: (path, line) -> [(byteoffset, text)]
    let mut subs: std::collections::HashMap<(String, i64), Vec<(usize, String)>> =
        std::collections::HashMap::new();
    for line in String::from_utf8_lossy(&b_out.stdout).lines() {
        // path:line:byteoffset:text  (grep -o prints one match per output line)
        if let Some((head, text)) = split3(line) {
            let (path, ln, byte) = head;
            subs.entry((path, ln)).or_default().push((byte, text));
        }
    }
    let mut matches: Vec<Value> = Vec::new();
    for line in String::from_utf8_lossy(&a_out.stdout).lines().take(200) {
        // path:line:content (grep -n, no -b → third field is the content)
        let (p1, rest) = match line.split_once(':') {
            Some(x) => x,
            None => continue,
        };
        let (ln, text) = match rest.split_once(':') {
            Some(x) => x,
            None => continue,
        };
        let ln: i64 = ln.parse().unwrap_or(0);
        let line_start = text.as_ptr() as usize; // unused; offsets come from pass B
        let _ = line_start;
        let rel = std::path::Path::new(p1)
            .strip_prefix(&base)
            .map(|r| r.to_string_lossy().to_string())
            .unwrap_or_else(|_| p1.to_string());
        // absolute_offset = byte offset of the line's start within the file
        // (freeze = ripgrep line start). Derive it from the first match's byte
        // offset minus its in-line start; grep -m/-n gives us line text only, so
        // compute the in-line start by locating the match text in the line.
        let line_subs = subs.get(&(p1.to_string(), ln)).cloned().unwrap_or_default();
        let mut submatch_json = Vec::new();
        let mut absolute_offset: i64 = 0;
        for (i, (byteoff, mtext)) in line_subs.iter().enumerate() {
            // in-line byte start = find mtext in the line (first occurrence for
            // the first sub; subsequent matches advance past prior ones)
            let from = submatch_json
                .last()
                .and_then(|v: &Value| v["end"].as_u64())
                .map(|e| e as usize + 1)
                .unwrap_or(0);
            let start = text[from.min(text.len())..]
                .find(mtext.as_str())
                .map(|p| p + from.min(text.len()))
                .unwrap_or(0);
            let end = start + mtext.len();
            if i == 0 {
                absolute_offset = (*byteoff as i64) - (start as i64);
            }
            submatch_json.push(json!({
                "match": {"text": mtext},
                "start": start,
                "end": end,
            }));
        }
        matches.push(json!({
            "path": {"text": rel},
            "lines": {"text": format!("{text}\n")},
            "line_number": ln,
            "absolute_offset": absolute_offset,
            "submatches": submatch_json,
        }));
    }
    Ok(Json(Value::Array(matches)))
}

/// Parse `path:line:byteoffset:text` (grep -o -b output; text may contain ':').
fn split3(line: &str) -> Option<((String, i64, usize), String)> {
    let (path, rest) = line.split_once(':')?;
    let (ln, rest) = rest.split_once(':')?;
    let (byte, text) = rest.split_once(':')?;
    Some((
        (path.to_string(), ln.parse().ok()?, byte.parse().ok()?),
        text.to_string(),
    ))
}

/// GET /question — all pending question requests (v1 Question.Request list;
/// oc-remote ConnectionService polls this while a dialog may be open).
async fn get_questions(State(st): State<Arc<AppState>>) -> Json<Value> {
    Json(Value::Array(st.question_gate.list()))
}

/// Validate a `que_`/`per_`-prefixed request id (freeze decode precedes the
/// 404/200: bad prefix → 400 Params).
pub(crate) fn validate_request_prefix(id: &str, prefix: &str) -> Result<(), HttpError> {
    if id.starts_with(prefix) {
        Ok(())
    } else {
        Err(HttpError::Params {
            message: format!(
                "Expected a string starting with \"{prefix}\", got \"{id}\"\n  at [\"requestID\"]"
            ),
        })
    }
}

/// Tagged QuestionNotFoundError (freeze v2 + v1 question routes).
pub(crate) fn question_not_found(id: &str) -> HttpError {
    HttpError::TaggedData {
        status: StatusCode::NOT_FOUND,
        tag: "QuestionNotFoundError",
        fields: json!({"requestID": id, "message": format!("Question request not found: {id}")}),
    }
}

/// POST /question/{id}/reply — {answers: [[labels]]} (v1 Reply payload) → true.
async fn post_question_reply(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, HttpError> {
    validate_request_prefix(&id, "que")?;
    let raw = body
        .get("answers")
        .and_then(|a| a.as_array())
        .ok_or_else(|| {
            HttpError::Api(ApiError {
                status: StatusCode::BAD_REQUEST,
                name: "BadRequest",
                message: "body must be {answers: [[String]]}".into(),
            })
        })?;
    let mut answers: Vec<Vec<String>> = Vec::with_capacity(raw.len());
    for a in raw {
        let row = a.as_array().ok_or_else(|| {
            HttpError::Api(ApiError {
                status: StatusCode::BAD_REQUEST,
                name: "BadRequest",
                message: "each answer must be an array of labels".into(),
            })
        })?;
        answers.push(
            row.iter()
                .map(|l| l.as_str().unwrap_or_default().to_string())
                .collect(),
        );
    }
    if st.question_gate.reply(&id, answers) {
        Ok(Json(json!(true)))
    } else {
        Err(question_not_found(&id))
    }
}

/// POST /question/{id}/reject — no body (oc-remote posts empty) → true.
async fn post_question_reject(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<Json<Value>, HttpError> {
    validate_request_prefix(&id, "que")?;
    if st.question_gate.reject(&id) {
        Ok(Json(json!(true)))
    } else {
        Err(question_not_found(&id))
    }
}

// ---- oc-remote contract family (Batch 4: command/shell) ----

/// v1 argsRegex tokenizer: quoted strings or non-space runs, then
/// quoteTrimRegex strips one leading/trailing quote per token.
pub fn split_command_args(arguments: &str) -> Vec<String> {
    let bytes = arguments.as_bytes();
    let mut out = Vec::new();
    let mut i = 0;
    while i < bytes.len() {
        let c = bytes[i];
        if c.is_ascii_whitespace() {
            i += 1;
            continue;
        }
        let (tok, next) = if c == b'"' || c == b'\'' {
            let quote = c;
            let mut j = i + 1;
            while j < bytes.len() && bytes[j] != quote {
                j += 1;
            }
            let end = (j + 1).min(bytes.len());
            (
                std::str::from_utf8(&bytes[i..end])
                    .unwrap_or("")
                    .to_string(),
                end,
            )
        } else {
            let mut j = i;
            while j < bytes.len() && !bytes[j].is_ascii_whitespace() {
                j += 1;
            }
            (
                std::str::from_utf8(&bytes[i..j]).unwrap_or("").to_string(),
                j,
            )
        };
        // quoteTrim: strip one leading then one trailing quote char
        let mut t = tok.as_str();
        if t.starts_with(['"', '\'']) {
            t = &t[1..];
        }
        if t.ends_with(['"', '\'']) && !t.is_empty() {
            t = &t[..t.len() - 1];
        }
        out.push(t.to_string());
        i = next;
    }
    out
}

/// v1 command template expansion (prompt.ts command()): $N placeholders (last
/// position swallows the remaining args), then $ARGUMENTS, then append rule.
/// `!`…`` shell snippets (ConfigMarkdown.shell) are NOT evaluated — neither
/// built-in template uses them (documented gap for exotic config commands).
pub fn expand_command(template: &str, arguments: &str) -> String {
    let args = split_command_args(arguments);
    // max placeholder position
    let mut last = 0usize;
    let b = template.as_bytes();
    let mut i = 0;
    while i + 1 < b.len() {
        if b[i] == b'$' && b[i + 1].is_ascii_digit() {
            let mut j = i + 1;
            while j < b.len() && b[j].is_ascii_digit() {
                j += 1;
            }
            if let Some(n) = std::str::from_utf8(&b[i + 1..j])
                .ok()
                .and_then(|d| d.parse::<usize>().ok())
            {
                last = last.max(n);
            }
            i = j;
        } else {
            i += 1;
        }
    }
    // replace $N
    let mut with_args = String::new();
    let mut i = 0;
    let mut had_placeholder = false;
    while i < b.len() {
        if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
            let mut j = i + 1;
            while j < b.len() && b[j].is_ascii_digit() {
                j += 1;
            }
            let n: usize = std::str::from_utf8(&b[i + 1..j])
                .unwrap_or("0")
                .parse()
                .unwrap_or(0);
            had_placeholder = true;
            let arg_index = n.saturating_sub(1);
            let value = if arg_index >= args.len() {
                String::new()
            } else if n == last {
                args[arg_index..].join(" ")
            } else {
                args[arg_index].clone()
            };
            with_args.push_str(&value);
            i = j;
        } else {
            // copy one utf-8 char
            let ch_len = {
                let ch = template[i..].chars().next().unwrap_or(' ');
                ch.len_utf8()
            };
            with_args.push_str(&template[i..i + ch_len]);
            i += ch_len;
        }
    }
    let uses_arguments = template.contains("$ARGUMENTS");
    let mut expanded = with_args.replace("$ARGUMENTS", arguments);
    if !had_placeholder && !uses_arguments && !arguments.trim().is_empty() {
        expanded = format!("{expanded}\n\n{arguments}");
    }
    expanded.trim().to_string()
}

/// v1 Session.Event.Error on unknown commands (oc-remote toast via
/// session.error SSE) + durable+s twin.
fn emit_session_error_event(st: &Arc<AppState>, sid: &str, message: &str) {
    let props = json!({
        "sessionID": sid,
        "error": {"name": "UnknownError", "data": {"message": message}},
    });
    if let Err(e) = ocserve_store::append_event(&st.writer, Some(sid), "session.error", &props) {
        tracing::warn!("session.error persist failed: {e:#}");
    }
    let seq = ocserve_store::next_event_seq(&st.db, sid).unwrap_or(0);
    let dir = st.paths["directory"].as_str().unwrap_or("/");
    st.bus.publish(ocserve_core::event::frame(
        dir,
        "session.error",
        props.clone(),
    ));
    st.bus.publish(ocserve_core::event::sync_frame(
        dir,
        "session.error",
        props,
        seq as u64,
        sid,
    ));
}

/// Busy rejection (v1 mapBusy → SessionBusyError): command/shell never queue
/// — they fail fast when the session is mid-prompt (upstream semantics).
pub(crate) fn session_busy(id: &str) -> ApiError {
    ApiError {
        status: StatusCode::CONFLICT,
        name: "SessionBusyError",
        message: format!("Session is busy: {id}"),
    }
}

/// POST /session/{id}/command — v1 promptSvc.command: look up the command
/// (built-ins init/review + config), expand the template, run the prompt
/// synchronously, return WithParts. Unknown → session.error SSE + 400.
async fn post_command(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    Json(payload): Json<Value>,
) -> Result<Json<Value>, HttpError> {
    // decode FIRST (HttpApi payload decode precedes handler — property order
    // messageID → agent → model → arguments → command, probe-pinned).
    let cmd = decode_v1(ocserve_core::wire::CommandRequest::decode(&payload))?;
    let exists_id = id.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })? {
        return Err(ApiError::not_found(format!("Session not found: {id}")).into());
    }
    let name = cmd.command.as_str();
    let arguments = cmd.arguments.as_str();
    // scoped: parking_lot guards are !Send — never held across run_prompt's
    // .await (Handler would stop accepting the future)
    let (template, available_owned): (Option<String>, Vec<String>) = {
        let payloads = st.payloads.read();
        let t = payloads
            .command
            .iter()
            .find(|c| c["name"].as_str() == Some(name))
            .and_then(|c| c["template"].as_str())
            .map(str::to_string);
        let names = payloads
            .command
            .iter()
            .filter_map(|c| c["name"].as_str().map(str::to_string))
            .collect();
        (t, names)
    };
    let available: Vec<&str> = available_owned.iter().map(String::as_str).collect();
    let Some(template) = template.as_deref() else {
        let msg = format!(
            "Command not found: \"{name}\".{}",
            if available.is_empty() {
                String::new()
            } else {
                format!(" Available commands: {}", available.join(", "))
            }
        );
        emit_session_error_event(&st, &id, &msg);
        return Err(ApiError {
            status: StatusCode::BAD_REQUEST,
            name: "BadRequest",
            message: msg,
        }
        .into());
    };
    let expanded = expand_command(template, arguments);
    let mut cmd_parts = json!([{ "type": "text", "text": expanded }]);
    // v1 parity: command.execute.before (prompt.ts:1461) — input
    // {command, sessionID, arguments}, output {parts} mutated before the
    // prompt runs. Fail-open (no sidecar/hook error → parts unchanged).
    let parts_out = ocserve_core::prompt::hook_mutate(
        st.plugins.get(),
        "command.execute.before",
        json!({
            "command": name,
            "sessionID": id,
            "arguments": arguments,
        }),
        json!({"parts": cmd_parts.clone()}),
    )
    .await;
    if let Some(p) = parts_out.get("parts").and_then(|p| p.as_array()) {
        cmd_parts = Value::Array(p.clone());
    }

    // Command wire `model` is a STRING (`providerID/modelID`) — parsed with
    // upstream `Provider.parseModel` semantics. The previous Value-path read
    // `/model/providerID` off the string and silently fell back to the
    // session/default model.
    let req = ocserve_core::wire::PromptRequest {
        message_id: cmd.message_id.clone(),
        model: cmd
            .model
            .as_deref()
            .map(ocserve_core::wire::parse_model_str),
        agent: cmd.agent.clone(),
        no_reply: false,
        tools: None,
        format: None,
        system: None,
        variant: cmd.variant.clone(),
        parts: cmd_parts.as_array().cloned().unwrap_or_default(),
    };
    let ctx = build_prompt_context(&st, &req, &id)?;
    let _release = lock_session(&st, &id).await?;
    let Ok(_guard) = _release.arc().try_lock() else {
        return Err(session_busy(&id).into());
    };
    let writer = st.writer.clone();
    let (info, parts) = ocserve_core::prompt::run_prompt_with(
        &ctx,
        &writer,
        &id,
        &req,
        ocserve_core::prompt::RunOpts {
            // K-TITLE: a command must never name the session after itself
            auto_title: false,
            ..Default::default()
        },
    )
    .await
    .map_err(prompt_err)?;
    drop(_guard);
    Ok(Json(json!({"info": info, "parts": parts})))
}

/// POST /session/{id}/shell — v1 promptSvc.shell: NO model. Runs the command
/// directly (bash), records synthetic-user + assistant/bash tool messages,
/// returns WithParts. Port note: shell.env plugin hook skipped (no configured
/// plugin subscribes); abort-of-running-shell not wired (documented).
async fn post_shell(
    State(st): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    Json(payload): Json<Value>,
) -> Result<Json<Value>, HttpError> {
    // decode FIRST: probe `{}` → `Missing key at ["agent"]` (freeze's
    // custom "command required" messages never existed upstream).
    let sh = decode_v1(ocserve_core::wire::ShellRequest::decode(&payload))?;
    let exists_id = id.clone();
    if !run_blocking(&st.db, move |db| {
        ocserve_store::session_exists(db, &exists_id)
    })
    .await
    .map_err(|e| ApiError {
        status: StatusCode::INTERNAL_SERVER_ERROR,
        name: "InternalError",
        message: format!("{e:#}"),
    })? {
        return Err(ApiError::not_found(format!("Session not found: {id}")).into());
    }
    let command = sh.command.as_str();
    let agent = sh.agent.as_str();
    let (pid, mid) = resolve_model(&st, sh.model.as_ref(), &id);

    let _release = lock_session(&st, &id).await?;
    let Ok(_guard) = _release.arc().try_lock() else {
        return Err(session_busy(&id).into());
    };
    let dir = st.paths["directory"].as_str().unwrap_or("/").to_string();

    // busy → run → idle (client spinner parity)
    let status_frame = |kind: &str| {
        ocserve_core::event::frame(
            &dir,
            "session.status",
            json!({"sessionID": id, "status": {"type": kind}}),
        )
    };
    st.bus.publish(status_frame("busy"));

    let started = now_ms_local();
    // client-supplied id honored like every other prompt-family route
    // (probe-unverified for shell specifically — field-probes.md).
    let user_msg_id = sh
        .message_id
        .clone()
        .unwrap_or_else(ocserve_core::ids::msg_id);
    let user_part = json!({
        "type": "text", "id": ocserve_core::ids::prt_id(),
        "sessionID": id, "messageID": user_msg_id,
        "text": "The following tool was executed by the user",
        "synthetic": true,
    });
    let user_info = json!({
        "id": user_msg_id,
        "sessionID": id,
        "role": "user",
        "agent": agent,
        "model": {"providerID": pid, "modelID": mid},
        "time": {"created": started},
        "summary": {"diffs": []},
    });
    let assistant_id = ocserve_core::ids::msg_id();
    let call_id = ocserve_core::ids::evt_id();
    let mut tool_part = json!({
        "type": "tool", "id": ocserve_core::ids::prt_id(),
        "sessionID": id, "messageID": assistant_id,
        "callID": call_id, "tool": "bash",
        "state": {
            "status": "running",
            "input": {"command": command},
            "time": {"start": started},
        },
    });

    // persist user msg + assistant msg, emit running state (upstream order)
    let writer = st.writer.clone();
    insert_message_http(
        &writer,
        &st,
        &id,
        &user_info,
        std::slice::from_ref(&user_part),
    )?;
    let assistant_info = json!({
        "id": assistant_id,
        "parentID": user_msg_id,
        "role": "assistant",
        "mode": agent,
        "agent": agent,
        "path": {"cwd": dir, "root": "/"},
        "cost": 0.0,
        "tokens": {"total": 0, "input": 0, "output": 0, "reasoning": 0,
                   "cache": {"write": 0, "read": 0}},
        "modelID": mid,
        "providerID": pid,
        "time": {"created": started, "completed": started},
        "finish": "stop",
        "id": assistant_id,
        "sessionID": id,
    });
    insert_message_http(&writer, &st, &id, &assistant_info, &[])?;
    emit_durable_http(
        &st,
        &id,
        "message.updated",
        json!({"sessionID": id, "info": user_info.clone()}),
    )?;
    emit_durable_http(
        &st,
        &id,
        "message.updated",
        json!({"sessionID": id, "info": assistant_info.clone()}),
    )?;
    emit_durable_http(
        &st,
        &id,
        "message.part.updated",
        json!({"sessionID": id, "part": user_part}),
    )?;
    emit_durable_http(
        &st,
        &id,
        "message.part.updated",
        json!({"sessionID": id, "part": tool_part.clone()}),
    )?;

    // execute directly (bash executor: stderr merged, output bounded,
    // default 120s cap — ShellInput carries no timeout). Off-worker: a long
    // shell command must not pin a tokio worker (K-EFFICIENCY).
    let exec = {
        let cmd = json!({"command": command});
        let d = dir.clone();
        tokio::task::spawn_blocking(move || {
            ocserve_tools::execute("bash", &cmd, std::path::Path::new(&d))
        })
        .await
        .map_err(|e| ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("tool join: {e}"),
        })?
    };
    let (status, output) = match exec {
        Ok(r) => ("completed", r.output),
        Err(e) => ("error", format!("Error: {e:#}")),
    };
    let completed = now_ms_local();
    // v1 finish() shape: completed always carries {title:"", output,
    // metadata:{output}} — no exit field (capture-verified)
    tool_part["state"] = json!({
        "status": status,
        "input": {"command": command},
        "output": output,
        "title": "",
        "metadata": {"output": ""},
        "time": {"start": started, "end": completed},
    });
    tool_part["state"]["metadata"]["output"] = tool_part["state"]["output"].clone();
    insert_message_http(
        &writer,
        &st,
        &id,
        &assistant_info,
        std::slice::from_ref(&tool_part),
    )?;
    emit_durable_http(
        &st,
        &id,
        "message.part.updated",
        json!({"sessionID": id, "part": tool_part.clone()}),
    )?;
    st.bus.publish(status_frame("idle"));

    Ok(Json(json!({"info": assistant_info, "parts": [tool_part]})))
}

fn now_ms_local() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

/// Persist a message via the store + emit its updated/part durable events.
fn insert_message_http(
    writer: &ocserve_store::Writer,
    st: &Arc<AppState>,
    sid: &str,
    info: &Value,
    parts: &[Value],
) -> Result<(), ApiError> {
    ocserve_store::insert_message(writer, Some(&*st.blobs), sid, info, parts).map_err(|e| {
        ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("{e:#}"),
        }
    })
}

/// Durable event from HTTP handlers (no PromptContext): persist + plain +
/// sync twin (seq from next_event_seq — single-flight per handler call).
/// Full session wire object merged with `overrides`, for a `session.updated`
/// event body. Falls back to `{id}` on a DB error so the caller never panics;
/// the event-shape guard (rule 19) flags any missing required key.
async fn session_wire_or_partial_http(st: &Arc<AppState>, sid: &str, overrides: Value) -> Value {
    let sid_owned = sid.to_string();
    run_blocking(&st.db, move |db| {
        ocserve_store::session_wire_with(db, &sid_owned, overrides)
    })
    .await
    .ok()
    .flatten()
    .unwrap_or_else(|| json!({"id": sid}))
}
pub(crate) fn emit_durable_http(
    st: &Arc<AppState>,
    sid: &str,
    event_type: &str,
    props: Value,
) -> Result<(), ApiError> {
    let props = ocserve_core::event::normalize_props(event_type, props);
    ocserve_store::append_event(&st.writer, Some(sid), event_type, &props).map_err(|e| {
        ApiError {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            name: "InternalError",
            message: format!("{e:#}"),
        }
    })?;
    let seq = ocserve_store::next_event_seq(&st.db, sid).unwrap_or(0);
    let dir = st.paths["directory"].as_str().unwrap_or("/");
    st.bus
        .publish(ocserve_core::event::frame(dir, event_type, props.clone()));
    st.bus.publish(ocserve_core::event::sync_frame(
        dir, event_type, props, seq as u64, sid,
    ));
    Ok(())
}

/// Freeze-shaped event id: `evt_` + 26 hex chars (upstream ids are time+random
/// 26-char alphanumerics; ours hash a nanos counter — clients only require
/// uniqueness, and byte-golden normalizes this span).
fn evt_id(kind: &str) -> String {
    use sha2::{Digest, Sha256};
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let mut h = Sha256::new();
    h.update(nanos.to_le_bytes());
    h.update(kind.as_bytes());
    let hex = hex::encode(h.finalize());
    format!("evt_{}", &hex[..26])
}

/// SSE endpoint. Freeze facts (byte-golden capture, PLAN §3):
/// - response headers: Cache-Control: no-cache, no-transform; X-Content-Type-Options: nosniff;
///   x-accel-buffering: no; Vary: Origin; chunked transfer
/// - NO `id:` lines, NO `retry:` lines
/// - first frame: server.connected WITHOUT directory/project wrapper
/// - then 10s JSON heartbeats
/// - frames wrapped: {"directory":..,"project":..,"payload":..} (except server.connected)
pub(crate) async fn global_event(State(st): State<Arc<AppState>>) -> Response {
    event_stream(st, EventFormat::Global).await
}

/// GET /event — v1 bare `Event` stream (`{id,type,properties}`, no wrapper).
/// The old wiring served the /global/event `{payload}` envelope here — a real
/// wire divergence (freeze /event: keys id/type/properties, no payload).
async fn bare_event(State(st): State<Arc<AppState>>) -> Response {
    event_stream(st, EventFormat::Bare).await
}

/// GET /api/event — v2 `V2Event` stream (`{id,type,data}`, properties folded
/// into `data`; `data` is the frozen V2 field name).
async fn v2_event(State(st): State<Arc<AppState>>) -> Response {
    event_stream(st, EventFormat::V2).await
}

#[derive(Clone, Copy)]
enum EventFormat {
    /// `{directory,project,payload:{id,type,properties}}`
    Global,
    /// `{id,type,properties}`
    Bare,
    /// `{id,type,data}`
    V2,
}

/// Rewrite one stored global-event frame into the requested wire format.
/// Returns None for frames the format does not carry (the sync twin is a
/// /global/event-only concern; /event and /api/event drop it — freeze emits
/// neither there).
fn encode_event_frame(stored: &str, fmt: EventFormat) -> Option<String> {
    if matches!(fmt, EventFormat::Global) {
        return Some(stored.to_string());
    }
    let v: Value = serde_json::from_str(stored).ok()?;
    let payload = v.get("payload")?;
    if payload.get("type").and_then(Value::as_str) == Some("sync") {
        return None;
    }
    let id = payload.get("id").cloned().unwrap_or(Value::Null);
    let ty = payload.get("type").cloned().unwrap_or(Value::Null);
    let props = payload.get("properties").cloned().unwrap_or(json!({}));
    Some(match fmt {
        EventFormat::Bare => json!({"id": id, "type": ty, "properties": props}).to_string(),
        EventFormat::V2 => json!({"id": id, "type": ty, "data": props}).to_string(),
        EventFormat::Global => unreachable!(),
    })
}

/// Build a synthetic `server.connected` / `server.heartbeat` frame in the given
/// format (freeze emits these with the format's own envelope: /global/event is
/// `{payload:{...}}` with NO directory/project on the connected frame, /event is
/// bare, /api/event is `{id,type,data}`).
fn synthetic_frame(fmt: EventFormat, event_type: &str) -> String {
    let id = evt_id(if event_type == "server.connected" {
        "connected"
    } else {
        "heartbeat"
    });
    match fmt {
        EventFormat::Global => json!({"payload": {
            "id": id, "type": event_type, "properties": {}}})
        .to_string(),
        EventFormat::Bare => json!({
            "id": id, "type": event_type, "properties": {}})
        .to_string(),
        EventFormat::V2 => json!({
            "id": id, "type": event_type, "data": {}})
        .to_string(),
    }
}

/// Shared SSE body for /event, /global/event, /api/event.
async fn event_stream(st: Arc<AppState>, fmt: EventFormat) -> Response {
    let connected = Event::default().data(synthetic_frame(fmt, "server.connected"));
    let rx = st.bus.subscribe();
    ocserve_metrics::gauge(
        "ocserve_sse_clients",
        st.bus.subscriber_count().max(1) as i64,
    );
    // The stream must own a bus clone: if the Router/state were dropped while
    // a subscriber still streams (oneshot tests), the Sender must survive.
    let bus_keepalive = st.bus.clone();
    // Drop-guard: decrement the client gauge when the stream ends (any exit).
    struct SseClientGuard;
    impl Drop for SseClientGuard {
        fn drop(&mut self) {
            ocserve_metrics::gauge_delta("ocserve_sse_clients", -1);
        }
    }
    let sse_guard = SseClientGuard;
    // Timeout-driven merge: heartbeat at exact 10s cadence regardless of
    // traffic; bus frames yielded as they arrive; Lagged/Closed ends the
    // stream (bounded-queue disconnect, PLAN §4).
    let events = stream::once(async move { Ok::<_, std::convert::Infallible>(connected) }).chain(
        stream::unfold(
            (
                rx,
                tokio::time::Instant::now() + Duration::from_secs(10),
                bus_keepalive,
                sse_guard,
                fmt,
            ),
            |(mut rx, mut next_hb, bus, guard, fmt)| async move {
                loop {
                    let dur = next_hb.saturating_duration_since(tokio::time::Instant::now());
                    match tokio::time::timeout(dur.max(Duration::from_millis(1)), rx.recv()).await {
                        Ok(Ok(frame)) => {
                            ocserve_metrics::counter("ocserve_sse_events_total", 1);
                            // frames this format does not carry (the sync twin on
                            // /event and /api/event) are skipped, not emitted as
                            // empty frames (freeze has no such frame).
                            if let Some(out) = encode_event_frame(&frame, fmt) {
                                return Some((
                                    Ok(Event::default().data(&out)),
                                    (rx, next_hb, bus, guard, fmt),
                                ));
                            }
                        }
                        Ok(Err(_lagged_or_closed)) => {
                            ocserve_metrics::labeled_counter(
                                "ocserve_event_ring_lag_total",
                                "reason=\"lagged_or_closed\"",
                                1,
                            );
                            return None;
                        }
                        Err(_elapsed) => {
                            let hb =
                                Event::default().data(synthetic_frame(fmt, "server.heartbeat"));
                            next_hb += Duration::from_secs(10);
                            return Some((Ok(hb), (rx, next_hb, bus, guard, fmt)));
                        }
                    }
                }
            },
        ),
    );
    // No axum KeepAlive: the freeze stream carries zero comment/retry frames
    // (PLAN §3 byte-golden); the 10s heartbeat payload is the only activity.
    let mut resp = Sse::new(events).into_response();
    // Freeze header set — axum's defaults differ (no-transform / nosniff / accel).
    let h = resp.headers_mut();
    for (k, v) in sse_expected_headers() {
        if let (Ok(name), Ok(val)) = (
            axum::http::HeaderName::try_from(k),
            HeaderValue::from_str(v),
        ) {
            h.insert(name, val);
        }
    }
    resp
}

/// Basic auth middleware (N6/PLAN §3: auth mode is a freeze artifact — live
/// instance is passwordless, so default is off; `basic` mode 401s everything
/// without matching credentials, including SSE and /metrics).
/// Per-request INFO logging (route hit-set capture) — opt-in via
/// OCSERVE_REQ_LOG=1 (read once at first request). Default off: at the
/// measured 10k+ rps a log line per request is pure journald amplification.
fn req_log() -> bool {
    static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
    *ON.get_or_init(|| {
        std::env::var("OCSERVE_REQ_LOG")
            .map(|v| v == "1")
            .unwrap_or(false)
    })
}

async fn auth_gate(
    State(st): State<Arc<AppState>>,
    req: axum::extract::Request,
    next: axum::middleware::Next,
) -> Result<Response, ApiError> {
    st.requests
        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    // SRE §2: request metrics (bounded route labels; measured at both exits)
    ocserve_metrics::counter("ocserve_requests_total", 1);
    let metric_t0 = std::time::Instant::now();
    let metric_label = format!(
        "route=\"{}\",method=\"{}\"",
        ocserve_metrics::route_label(req.uri().path()),
        req.method()
    );
    let record = move |t0: std::time::Instant| {
        ocserve_metrics::observe(
            "ocserve_http_request_duration_seconds",
            &metric_label,
            t0.elapsed().as_micros() as u64,
        );
    };
    // Hit-set capture (M1 §PLAN C4): route-inventory probes set
    // OCSERVE_REQ_LOG=1. Default OFF — one INFO line per request is pure
    // journald write amplification at the measured 10k+ rps (and the live
    // journal showed 1400 /api/health lines in hours of idle traffic).
    if req_log() {
        tracing::info!("req {} {}", req.method(), req.uri().path());
    }
    if let Some((user, pass)) = &st.auth {
        let authorized = req
            .headers()
            .get(axum::http::header::AUTHORIZATION)
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.strip_prefix("Basic "))
            .and_then(|b64| {
                use base64::Engine;
                base64::engine::general_purpose::STANDARD.decode(b64).ok()
            })
            .map(|raw| {
                let text = String::from_utf8_lossy(&raw);
                text == format!("{user}:{pass}")
            })
            .unwrap_or(false);
        if !authorized {
            record(metric_t0);
            return Err(ApiError {
                status: StatusCode::UNAUTHORIZED,
                name: "UnauthorizedError",
                message: "unauthorized".into(),
            });
        }
    }
    let resp = next.run(req).await;
    record(metric_t0);
    Ok(resp)
}

/// Router for the P0 surface. Auth is an AppState decision (freeze §3).
pub fn router(state: Arc<AppState>) -> Router {
    Router::new()
        .route("/global/health", get(health))
        .route("/doc", get(get_doc))
        .route("/config", get(get_config))
        .route("/config/providers", get(get_config_providers))
        .route("/provider", get(get_provider))
        .route("/agent", get(get_agent))
        .route("/command", get(get_command))
        .route("/experimental/console", get(get_console))
        .route("/experimental/tool", get(get_experimental_tool))
        .route("/experimental/tool/ids", get(get_experimental_tool_ids))
        .route("/experimental/capabilities", get(get_capabilities))
        .route("/path", get(get_path))
        .route("/project", get(get_projects))
        .route("/project/current", get(get_project_current))
        .route("/session", get(get_sessions).post(post_session))
        .route("/session/{id}/fork", axum::routing::post(post_fork))
        .route("/session/status", get(session_status))
        .route("/session/search", axum::routing::post(search_messages))
        .route("/mcp/{name}/{action}", axum::routing::post(mcp_action))
        .route("/mcp/{name}/auth", axum::routing::delete(mcp_auth_remove))
        .route(
            "/auth/{providerID}",
            axum::routing::put(auth_put).delete(auth_delete),
        )
        .route("/provider/auth", axum::routing::get(provider_auth_methods))
        .route("/global/dispose", axum::routing::post(global_dispose))
        .route("/config", axum::routing::patch(patch_config))
        .route("/global/config", axum::routing::patch(patch_config))
        .route(
            "/session/{id}/summarize",
            axum::routing::post(post_summarize),
        )
        .route(
            "/session/{id}",
            get(get_session)
                .patch(patch_session)
                .delete(delete_session_route),
        )
        .route(
            "/session/{id}/message",
            get(get_messages).post(post_message),
        )
        .route("/global/event", get(global_event))
        // Distinct frozen shapes (live-probed 2026-10-11): /global/event wraps
        // `{directory,project,payload:{...}}`; /event is bare `{id,type,properties}`;
        // /api/event is v2 `{id,type,data}`. Serving one shape from all three was
        // a real wire divergence.
        .route("/event", get(bare_event))
        .route("/api/event", get(v2_event))
        .route("/metrics", get(metrics))
        // TUI-attach probes (captured live; PLAN §2 hit-set expansion).
        // The /api/* routes are registered in the v2 block below.
        .route("/experimental/resource", get(experimental_resource))
        .route(
            "/experimental/workspace",
            get(experimental_workspace).post(v2::workspace_create),
        )
        // TUI attach calls this at boot (observed live 2026-10-08); freeze
        // answers [] with 200 under an empty workspace state.
        .route(
            "/experimental/workspace/status",
            get(experimental_workspace_status),
        )
        .route("/formatter", get(formatter_list))
        .route("/lsp", get(lsp_list))
        .route("/project/{id}/directories", get(project_directories))
        .route("/vcs", get(vcs_info))
        .route("/vcs/status", get(vcs_status))
        .route("/vcs/diff", get(vcs_diff))
        .route("/vcs/diff/raw", get(vcs_diff_raw))
        .route("/file/status", get(file_status))
        .route("/find/symbol", get(find_symbol))
        .route("/global/config", get(global_config_get))
        .route(
            "/session/{id}/prompt_async",
            axum::routing::post(post_prompt_async),
        )
        .route("/session/{id}/children", get(get_children))
        .route("/file", get(list_directory))
        .route("/file/content", get(read_file_content))
        .route("/find/file", get(find_files))
        .route("/find", get(find_text))
        .route("/question", get(get_questions))
        .route(
            "/question/{id}/reply",
            axum::routing::post(post_question_reply),
        )
        .route(
            "/question/{id}/reject",
            axum::routing::post(post_question_reject),
        )
        .route("/session/{id}/todo", get(get_todos))
        .route("/session/{id}/abort", axum::routing::post(post_abort))
        .route("/session/{id}/command", axum::routing::post(post_command))
        .route("/session/{id}/shell", axum::routing::post(post_shell))
        .route(
            "/session/{id}/message/{mid}",
            get(get_message_by_id).delete(delete_message_route),
        )
        // legacy permission respond (deprecated upstream; oc-remote uses
        // /permission/{id}/reply — this is the SDK/v1 shape)
        .route(
            "/session/{id}/permissions/{pid}",
            axum::routing::post(post_session_permission_respond),
        )
        // v1 control-plane log writer (web UI diagnostics)
        .route("/log", axum::routing::post(post_log))
        // revert/unrevert (SNAPSHOT.md): real file-restoring revert — the
        // snapshot engine restores pre-turn files, reverts tool patch parts,
        // computes the range diff, sets the marker. Busy → 409.
        .route("/session/{id}/revert", axum::routing::post(post_revert))
        .route("/session/{id}/unrevert", axum::routing::post(post_unrevert))
        // /instance/dispose — v1 single-instance dispose (freeze: true)
        .route("/instance/dispose", axum::routing::post(instance_dispose))
        // /session/{id}/init — built-in `init` command (guided AGENTS.md;
        // payload {messageID,providerID,modelID} — probed freeze)
        .route("/session/{id}/init", axum::routing::post(post_init))
        // provider OAuth: ocserve implements no OAuth flows; the route
        // exists and answers the freeze shape for non-OAuth providers
        // (400 BadRequest {}) instead of falling through to the SPA.
        .route(
            "/provider/{id}/oauth/authorize",
            axum::routing::post(post_provider_oauth),
        )
        .route(
            "/provider/{id}/oauth/callback",
            axum::routing::post(post_provider_oauth_callback),
        )
        // MCP OAuth: tagged shapes probed (unknown → 404
        // McpServerNotFoundError; known/non-OAuth → 400
        // McpUnsupportedOAuthError). No OAuth-capable servers in practice.
        .route(
            "/mcp/{name}/auth/authenticate",
            axum::routing::post(post_mcp_auth_authenticate),
        )
        .route(
            "/mcp/{name}/auth/callback",
            axum::routing::post(post_mcp_auth_callback),
        )
        .route(
            "/session/{id}/message/{mid}/part/{pid}",
            axum::routing::delete(delete_part_route).patch(patch_part_route),
        )
        .route("/experimental/session", get(get_experimental_sessions))
        // ---- /pty/* (web-UI terminal; freeze-probed 2026-10-08) ----
        .route("/pty/shells", get(pty::get_shells))
        .route("/pty", get(pty::list).post(pty::create))
        .route(
            "/pty/{id}",
            get(pty::get).put(pty::update).delete(pty::remove),
        )
        .route(
            "/pty/{id}/connect-token",
            axum::routing::post(pty::connect_token),
        )
        .route("/pty/{id}/connect", get(pty::connect))
        // /tui/* — external-controller ingress (PLAN §17; VS Code extension)
        .route(
            "/tui/append-prompt",
            axum::routing::post(tui::append_prompt),
        )
        .route("/tui/open-help", axum::routing::post(tui::open_help))
        .route(
            "/tui/open-sessions",
            axum::routing::post(tui::open_sessions),
        )
        .route("/tui/open-themes", axum::routing::post(tui::open_themes))
        .route("/tui/open-models", axum::routing::post(tui::open_models))
        .route(
            "/tui/submit-prompt",
            axum::routing::post(tui::submit_prompt),
        )
        .route("/tui/clear-prompt", axum::routing::post(tui::clear_prompt))
        .route(
            "/tui/execute-command",
            axum::routing::post(tui::execute_command),
        )
        .route("/tui/show-toast", axum::routing::post(tui::show_toast))
        .route("/tui/publish", axum::routing::post(tui::publish_event))
        .route(
            "/tui/select-session",
            axum::routing::post(tui::select_session),
        )
        .route("/tui/control/next", get(tui::control_next))
        .route(
            "/tui/control/response",
            axum::routing::post(tui::control_response),
        )
        .route("/mcp", get(get_mcp).post(post_mcp))
        .route("/permission", get(get_permissions))
        .route(
            "/permission/{id}/reply",
            axum::routing::post(post_permission_reply),
        )
        // ---- v2 `/api/*` surface (1.18.31 TUI + web UI DataProvider) ----
        // Location-scoped collections: {location, data}.
        .route("/api/health", get(v2::health))
        .route("/api/location", get(v2::location))
        .route("/api/agent", get(get_api_agent))
        .route("/api/model", get(v2::list_models))
        .route("/api/provider", get(v2::list_providers))
        .route("/api/provider/{providerID}", get(v2::get_provider))
        .route("/api/skill", get(v2::list_skills))
        .route("/skill", get(v2::list_skills_raw))
        .route("/vcs/apply", axum::routing::post(v2::vcs_apply))
        .route("/api/command", get(get_api_command))
        .route("/api/integration", get(v2::list_integrations))
        .route("/api/integration/{integrationID}", get(v2::get_integration))
        .route("/api/reference", get(v2::list_references))
        .route("/api/fs/find", get(v2::fs_find))
        .route("/api/fs/list", get(v2::fs_list))
        .route("/api/fs/read/{*path}", get(v2::fs_read))
        // Sessions.
        .route(
            "/api/session",
            get(v2::list_sessions).post(v2::create_session),
        )
        .route("/api/session/active", get(v2::active_sessions))
        .route("/api/session/{sessionID}", get(v2::get_session))
        .route("/api/session/{sessionID}/message", get(v2::list_messages))
        .route(
            "/api/session/{sessionID}/message/{messageID}",
            get(v2::get_message),
        )
        .route("/api/session/{sessionID}/context", get(v2::session_context))
        .route("/api/session/{sessionID}/history", get(v2::session_history))
        .route(
            "/api/session/{sessionID}/agent",
            axum::routing::post(v2::switch_agent),
        )
        .route(
            "/api/session/{sessionID}/model",
            axum::routing::post(v2::switch_model),
        )
        .route(
            "/api/session/{sessionID}/prompt",
            axum::routing::post(v2::prompt),
        )
        .route(
            "/api/session/{sessionID}/interrupt",
            axum::routing::post(v2::interrupt),
        )
        .route(
            "/api/session/{sessionID}/compact",
            axum::routing::post(v2::compact),
        )
        .route(
            "/api/session/{sessionID}/wait",
            axum::routing::post(v2::wait),
        )
        .route(
            "/api/session/{sessionID}/revert/stage",
            axum::routing::post(v2::revert_stage),
        )
        .route(
            "/api/session/{sessionID}/revert/clear",
            axum::routing::post(v2::revert_clear),
        )
        .route(
            "/api/session/{sessionID}/revert/commit",
            axum::routing::post(v2::revert_commit),
        )
        .route(
            "/api/session/{sessionID}/permission/{requestID}/reply",
            axum::routing::post(v2::permission_reply),
        )
        .route(
            "/api/session/{sessionID}/question/{requestID}/reply",
            axum::routing::post(v2::question_reply),
        )
        .route(
            "/api/session/{sessionID}/question/{requestID}/reject",
            axum::routing::post(v2::question_reject),
        )
        // Permissions / questions.
        .route("/api/permission/saved", get(v2::permission_saved))
        .route(
            "/api/permission/saved/{id}",
            axum::routing::delete(v2::permission_saved_delete),
        )
        .route("/api/permission/request", get(v2::permission_request))
        .route("/api/question/request", get(v2::question_request))
        .route(
            "/api/session/{sessionID}/question",
            get(v2::session_question),
        )
        .route(
            "/api/session/{sessionID}/permission/{requestID}",
            get(v2::session_permission_one),
        )
        .route(
            "/api/session/{sessionID}/permission",
            get(v2::session_permission).post(v2::permission_create),
        )
        // pty under /api (same implementation).
        .route("/api/pty", get(v2::pty_list).post(v2::pty_create))
        .route(
            "/api/pty/{ptyID}",
            get(v2::pty_get).put(v2::pty_update).delete(v2::pty_remove),
        )
        .route(
            "/api/pty/{ptyID}/connect-token",
            axum::routing::post(v2::pty_connect_token),
        )
        .route("/api/pty/{ptyID}/connect", get(pty::connect))
        // v1 path the TUI session view calls (SidebarFiles crash path).
        .route("/session/{sessionID}/diff", get(v2::session_diff))
        .route(
            "/session/{sessionID}/share",
            axum::routing::post(v2::session_share).delete(v2::session_unshare),
        )
        // Experimental action stubs (freeze-probed shapes).
        .route("/experimental/console/orgs", get(v2::console_orgs))
        .route(
            "/experimental/console/switch",
            axum::routing::post(v2::console_switch),
        )
        .route(
            "/experimental/workspace/adapter",
            get(v2::workspace_adapter),
        )
        .route(
            "/experimental/workspace/sync-list",
            axum::routing::post(v2::workspace_sync_list),
        )
        .route(
            "/experimental/workspace/warp",
            axum::routing::post(v2::workspace_warp),
        )
        .route(
            "/experimental/worktree",
            get(v2::worktree_list).post(v2::worktree_create),
        )
        .route(
            "/experimental/worktree",
            axum::routing::delete(v2::worktree_remove),
        )
        .route(
            "/experimental/worktree/reset",
            axum::routing::post(v2::worktree_reset),
        )
        .route("/api/session/{sessionID}/event", get(v2::session_event))
        .route(
            "/project/{projectID}",
            axum::routing::patch(v2::patch_project),
        )
        // Sync / integration / credential / upgrade (mostly cited or
        // no-backend; freeze-probed shapes).
        .route("/sync/start", axum::routing::post(v2::sync_start))
        .route("/sync/replay", axum::routing::post(v2::sync_replay))
        .route("/sync/history", axum::routing::post(v2::sync_history))
        .route("/sync/steal", axum::routing::post(v2::sync_steal))
        .route("/project/git/init", axum::routing::post(v2::git_init))
        .route("/global/upgrade", axum::routing::post(v2::global_upgrade))
        .route(
            "/api/integration/{integrationID}/connect/key",
            axum::routing::post(v2::integration_connect_key),
        )
        .route(
            "/api/integration/{integrationID}/connect/oauth",
            axum::routing::post(v2::integration_connect_oauth),
        )
        .route(
            "/api/integration/attempt/{attemptID}",
            get(v2::integration_attempt_get).delete(v2::integration_attempt_delete),
        )
        .route(
            "/api/integration/attempt/{attemptID}/complete",
            axum::routing::post(v2::integration_attempt_complete),
        )
        .route(
            "/api/credential/{credentialID}",
            axum::routing::patch(v2::credential_update).delete(v2::credential_remove),
        )
        .route(
            "/experimental/control-plane/move-session",
            axum::routing::post(v2::control_move_session),
        )
        .route(
            "/experimental/project/{projectID}/copy",
            axum::routing::post(v2::project_copy_create).delete(v2::project_copy_remove),
        )
        .route(
            "/experimental/project/{projectID}/copy/generate-name",
            axum::routing::post(v2::project_copy_generate_name),
        )
        .route(
            "/experimental/project/{projectID}/copy/refresh",
            axum::routing::post(v2::project_copy_refresh),
        )
        .route(
            "/experimental/session/{sessionID}/background",
            axum::routing::post(v2::session_background),
        )
        .route(
            "/experimental/workspace/{id}",
            axum::routing::delete(v2::workspace_delete),
        )
        .layer(axum::middleware::from_fn_with_state(
            state.clone(),
            auth_gate,
        ))
        // CORS (upstream cors.ts parity) sits OUTSIDE auth: preflights must
        // answer without credentials, exactly like upstream.
        .layer(axum::middleware::from_fn_with_state(
            state.clone(),
            crate::ui::cors_gate,
        ))
        // Catch-all: unmatched paths serve the web UI (embedded upstream-side
        // or proxied to app.opencode.ai), matching v1 serveUIEffect. Disabled
        // via OCSERVE_UI=0 → JSON 404.
        .fallback(crate::ui::ui_fallback)
        // Upstream's catch-all is a *route*, not a fallback: a known path with
        // an unknown method (e.g. PATCH /pty/{id}) also gets the SPA, not a
        // 405. Probed against freeze 2026-10-08.
        .method_not_allowed_fallback(crate::ui::ui_fallback)
        .with_state(state)
}

/// CORS extras: OCSERVE_CORS (comma-separated) — the `--cors` equivalent.
/// Config-file `server.cors` support arrives with the config passthrough;
/// env is the runtime-tunable equivalent until then.
fn read_cors_env() -> Vec<String> {
    std::env::var("OCSERVE_CORS")
        .ok()
        .map(|v| {
            v.split(',')
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .map(str::to_string)
                .collect()
        })
        .unwrap_or_default()
}

/// JSON key-path projection: `$`, `$.a`, `$.a[].b`, … (values ignored).
/// Shared oracle for keys-mode golden comparisons (manifest mode=keys).
pub fn keypaths(v: &serde_json::Value, prefix: &str, out: &mut std::collections::BTreeSet<String>) {
    match v {
        serde_json::Value::Object(m) => {
            for (k, val) in m {
                let p = format!("{prefix}.{k}");
                out.insert(p.clone());
                keypaths(val, &p, out);
            }
        }
        serde_json::Value::Array(a) => {
            let p = format!("{prefix}[]");
            if let Some(first) = a.first() {
                keypaths(first, &p, out);
            } else {
                out.insert(p);
            }
        }
        _ => {}
    }
}

/// Header values the SSE route must carry (asserted by tests against golden bytes).
pub fn sse_expected_headers() -> Vec<(&'static str, &'static str)> {
    vec![
        ("cache-control", "no-cache, no-transform"),
        ("x-content-type-options", "nosniff"),
        ("content-type", "text/event-stream"),
        ("x-accel-buffering", "no"),
    ]
}

#[allow(dead_code)]
fn _assert_header_value(v: &str) -> HeaderValue {
    HeaderValue::from_str(v).expect("static header value")
}

#[cfg(test)]
mod limit_matrix {
    use super::parse_limit;

    #[test]
    fn matches_probe_envelopes() {
        // Byte strings from message_page_contract.json probes (upstream 1.18.31)
        assert_eq!(
            parse_limit("abc").unwrap_err(),
            "Expected an integer, got NaN\n  at [\"limit\"]"
        );
        assert_eq!(
            parse_limit("5.5").unwrap_err(),
            "Expected an integer, got 5.5\n  at [\"limit\"]"
        );
        assert_eq!(
            parse_limit("-0.5").unwrap_err(),
            "Expected an integer, got -0.5\n  at [\"limit\"]"
        );
        assert_eq!(
            parse_limit("-1").unwrap_err(),
            "Expected a value greater than or equal to 0, got -1\n  at [\"limit\"]"
        );
        assert_eq!(
            parse_limit("999999999999999999999").unwrap_err(),
            "Expected an integer, got 1e+21\n  at [\"limit\"]"
        );
        assert_eq!(
            parse_limit("Infinity").unwrap_err(),
            "Expected an integer, got Infinity\n  at [\"limit\"]"
        );
    }

    #[test]
    fn accepts_integers_like_js_number() {
        assert_eq!(parse_limit("+5"), Ok(5));
        assert_eq!(parse_limit("1e3"), Ok(1000)); // JS Number("1e3") = 1000
        assert_eq!(parse_limit("0"), Ok(0));
        assert_eq!(parse_limit("9007199254740991"), Ok(9007199254740991)); // 2^53-1
    }
}

#[cfg(test)]
mod f1_blocking_tests {
    use super::{AppState, Payloads, router, run_blocking};
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::Duration;

    /// PERF-10X F1: a blocking store call must NOT park the async worker.
    /// Single-worker runtime makes this discriminating: with the call inlined
    /// (the pre-F1 shape) the ticker cannot advance while the call runs
    /// (after == before → red); via the blocking pool it keeps ticking.
    /// M3: `GET /find/file`'s directory walk must execute OFF the tokio
    /// worker. Structural proof via thread identity, not timing: a ticker-
    /// based version of this test passed with the walk still inline (a few
    /// thousand files walk in microseconds), so it proved nothing.
    ///
    /// The probe drives the REAL route — calling `walk_files` directly would
    /// bypass the very `spawn_blocking` under test, which the first version
    /// of this test did (and it failed for exactly that reason).
    #[tokio::test]
    async fn find_files_walk_runs_off_the_worker() {
        use tower::ServiceExt as _;
        let dir = std::env::temp_dir().join(format!(
            "ocserve-m3-walk-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        for i in 0..200 {
            std::fs::write(dir.join(format!("f{i}.txt")), b"x").unwrap();
        }

        let mut payloads = Payloads::default();
        payloads.config["directory"] = serde_json::Value::String(dir.to_string_lossy().to_string());
        let st = AppState::with_payloads(None, payloads);

        let handler_thread = std::thread::current().id();
        let req = axum::http::Request::builder()
            .uri("/find/file?query=f1&limit=10")
            .body(axum::body::Body::empty())
            .unwrap();
        let resp = tokio::spawn(async move { router(st).oneshot(req).await })
            .await
            .expect("route task")
            .expect("infallible");
        assert_eq!(resp.status(), 200);

        let walked_on = super::WALK_THREAD.lock().ok().and_then(|g| *g);
        let walked_on = walked_on.expect("walk_files must record its thread");
        assert_ne!(
            walked_on, handler_thread,
            "the walk ran on the async worker — blocking syscalls must go \
             through spawn_blocking (M3)"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// PERF-10X F1: a blocking store call must NOT park the async worker.
    /// Single-worker runtime makes this discriminating: with the call inlined
    /// (the pre-F1 shape) the ticker cannot advance while the call runs
    /// (after == before → red); via the blocking pool it keeps ticking.
    /// Negative control: swap run_blocking for the inline sleep → red.
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]

    async fn run_blocking_frees_the_worker() {
        let ticks = Arc::new(AtomicU64::new(0));
        let t = ticks.clone();
        let ticker = tokio::spawn(async move {
            let mut i = 0u64;
            loop {
                tokio::time::sleep(Duration::from_millis(10)).await;
                i += 1;
                t.store(i, Ordering::Relaxed);
            }
        });
        tokio::time::sleep(Duration::from_millis(30)).await;
        let before = ticks.load(Ordering::Relaxed);
        assert!(before > 0, "ticker must be running before the probe");

        // The probe runs as a SPAWNED task: only spawned tasks execute on
        // worker threads (the #[tokio::test] body itself runs on block_on's
        // thread — an inline sleep there parked nothing and the first version
        // of this test passed its own negative control; caught 2026-10-06).
        let probe = tokio::spawn(async move {
            run_blocking(std::path::Path::new("/nonexistent/f1-probe"), |_db| {
                std::thread::sleep(Duration::from_millis(250));
                Ok::<(), anyhow::Error>(())
            })
            .await
            .expect("probe op");
        });
        probe.await.expect("probe join");

        let after = ticks.load(Ordering::Relaxed);
        ticker.abort();
        assert!(
            after > before,
            "blocking call parked the worker: before={before} after={after}"
        );
    }
}

#[cfg(test)]
mod f5_wire {
    use super::*;

    /// Identity control: cached bytes == what axum's Json would produce
    /// (both serde_json::to_vec) — one value per key, no drift.
    #[test]
    fn wire_bytes_are_serde_identical() {
        let st = AppState::new();
        let p = st.payloads.read();
        let w = st.wire.read();
        for (k, v) in [
            ("config", &p.config),
            ("config_providers", &p.config_providers),
            ("provider", &p.provider),
            ("console", &p.console),
            ("capabilities", &p.capabilities),
        ] {
            let cached = w.get(k).unwrap_or_else(|| panic!("missing wire key {k}"));
            let want = serde_json::to_vec(v).unwrap();
            assert_eq!(
                &cached.identity[..],
                &want[..],
                "wire bytes for {k} must equal Value serialization"
            );
        }
        for (k, v) in [("agent", &p.agent), ("command", &p.command)] {
            let cached = w.get(k).unwrap_or_else(|| panic!("missing wire key {k}"));
            let want = serde_json::to_vec(v).unwrap();
            assert_eq!(&cached.identity[..], &want[..]);
        }
    }

    /// Kill-switch: wire_off => no cache, wire_json falls back (None).
    #[test]
    fn wire_off_disables_cache() {
        let empty = rebuild_wire(
            &Payloads::default(),
            true,
            &json!({"directory": "/", "project": {"id": "global", "directory": "/"}}),
        );
        assert!(empty.is_empty(), "off => nothing serialized");
        let st = AppState::new();
        st.wire_off
            .store(true, std::sync::atomic::Ordering::Relaxed);
        assert!(wire_json(&st, "config").is_none(), "off => Value path");
    }

    /// v2 metadata wire entries: cached bytes == the pure builder serialized
    /// (the single-source-of-truth parity — a handler drift from its builder
    /// would serve stale shapes to browsers with `Accept-Encoding`).
    #[test]
    fn v2_metadata_wire_matches_builders() {
        for (catalog, name) in [
            (
                serde_json::json!({"deepseek": {"name": "DeepSeek"}}),
                "with-catalog",
            ),
            (serde_json::json!({}), "empty-catalog"),
        ] {
            let payloads = Payloads {
                provider: serde_json::json!({
                    "connected": ["opencode", "deepseek"],
                    "all": [
                        {"id": "opencode", "name": "opencode", "models": {"m": {"id": "m", "providerID": "opencode", "name": "M"}}},
                        {"id": "deepseek", "name": "DeepSeek", "models": {"d": {"id": "d", "providerID": "deepseek", "name": "D"}}},
                    ]
                }),
                config: serde_json::json!({"provider": {"entrim": {"name": "Entrim"}}}),
                catalog: std::sync::Arc::new(catalog),
                ..Default::default()
            };
            let location = json!({"directory": "/", "project": {"id": "global", "directory": "/"}});
            let wire = rebuild_wire(&payloads, false, &location);
            for (key, want) in [
                ("api_model", crate::v2::models_body(&payloads, &location)),
                (
                    "api_provider",
                    crate::v2::providers_body(&payloads, &location),
                ),
                (
                    "api_integration",
                    crate::v2::integrations_body(&payloads, &location),
                ),
            ] {
                let e = wire
                    .get(key)
                    .unwrap_or_else(|| panic!("[{name}] missing wire key {key}"));
                assert_eq!(
                    &e.identity[..],
                    &serde_json::to_vec(&want).unwrap()[..],
                    "[{name}] {key} wire bytes must equal the pure builder"
                );
            }
        }
    }
}

#[cfg(test)]
mod doc_contract {
    /// P0: `/doc` serves the vendored OpenAPI contract byte-for-byte; it must
    /// be valid JSON with the expected shape (162 paths / 188 ops). This is the
    /// oracle P4 validation and the P1 coverage guard read.
    #[test]
    fn spec_is_valid_and_complete() {
        let d: serde_json::Value =
            serde_json::from_str(crate::doc::SPEC_JSON).expect("vendored spec must be valid JSON");
        assert_eq!(d["openapi"], "3.1.0");
        let paths = d["paths"].as_object().expect("paths object");
        assert_eq!(paths.len(), 162, "frozen 1.18.31 has 162 paths");
        let ops: usize = paths
            .values()
            .map(|v| {
                v.as_object()
                    .unwrap()
                    .keys()
                    .filter(|k| matches!(k.as_str(), "get" | "post" | "put" | "delete" | "patch"))
                    .count()
            })
            .sum();
        assert_eq!(ops, 188, "frozen 1.18.31 has 188 operations");
        assert!(d["components"]["schemas"].as_object().unwrap().len() >= 400);
    }
}