lex_api/

handlers.rs

//! Request routing for the agent API.
//!
//! Each handler is a synchronous function that returns
//! `Result<serde_json::Value, ApiError>`. The dispatcher wraps the result
//! in an HTTP response — successes as 200 with the JSON body, structured
//! errors as 4xx/5xx with a JSON envelope.

use indexmap::IndexMap;
use lex_ast::canonicalize_program;
use lex_bytecode::{compile_program, vm::Vm, Value};
use lex_runtime::{check_program as check_policy, DefaultHandler, Policy};
use lex_store::Store;
use crate::publish_examples::record_examples_for_publish;
use lex_syntax::{load_package, load_program_from_str, Manifest};
use lex_vcs::{MergeSession, MergeSessionId};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};
use tiny_http::{Header, Method, Request, Response};

/// The function declarations in a canonicalized program, by name.
fn stage_fns(stages: &[lex_ast::Stage]) -> BTreeMap<String, lex_ast::FnDecl> {
    stages.iter().filter_map(|s| match s {
        lex_ast::Stage::FnDecl(fd) => Some((fd.name.clone(), fd.clone())),
        _ => None,
    }).collect()
}

/// The type declarations in a canonicalized program, by name — so the
/// publish diff captures `type`s alongside functions (#895).
fn stage_types(stages: &[lex_ast::Stage]) -> BTreeMap<String, lex_ast::TypeDecl> {
    stages.iter().filter_map(|s| match s {
        lex_ast::Stage::TypeDecl(td) => Some((td.name.clone(), td.clone())),
        _ => None,
    }).collect()
}

pub struct State {
    pub store: Mutex<Store>,
    /// Filesystem root of the store. Held alongside the `Store`
    /// itself so handlers that need to read store-level files
    /// (e.g. `users.json` for actor auth) don't have to round-
    /// trip through the lock.
    pub root: PathBuf,
    /// In-memory merge sessions, keyed by MergeSessionId. Sessions
    /// are ephemeral by design (#134 foundation): they live for the
    /// lifetime of the server process and are GC'd on commit. A
    /// future slice can persist them to disk so a session survives
    /// process restarts. For now an agent that gets unlucky with a
    /// restart re-runs `merge/start` and gets a fresh session.
    pub sessions: Mutex<HashMap<MergeSessionId, ApiMergeSession>>,
    /// Optional server-imposed ceiling on the effect policy honored
    /// by `/v1/run` and `/v1/replay`. `None` (the default, used by
    /// single-tenant `lex serve`) runs the caller's request policy
    /// as-is — the operator *is* the caller there, so that's
    /// intended. When `Some`, the request policy is clamped via
    /// [`clamp_policy`] so it can only *narrow* the ceiling, never
    /// widen it.
    ///
    /// Any embedder that exposes this API to untrusted callers — a
    /// hosted, multi-tenant gateway like lex-hub — MUST set this.
    /// Without it the request body can grant itself `[proc]`
    /// (arbitrary subprocess spawn), `[fs_*]` over `/`, and
    /// unrestricted `[net]`: arbitrary code execution as the server
    /// process. See lex-hub#6.
    ///
    /// NOTE: an empty scope list means "any path/host" in the
    /// runtime, so a ceiling that puts `fs_read`/`fs_write`/`net` in
    /// `allow_effects` MUST also populate the matching scope list
    /// (`allow_fs_read`, …) or it re-opens the wildcard. Granting
    /// none of those kinds is the safe default.
    pub policy_ceiling: Option<Policy>,
    /// Per-head op-history indexes and paged deltas behind
    /// `/v1/ops/since` (#971), so a paged pull walks the op log once
    /// instead of once per page. See [`crate::ops_since_http`].
    pub(crate) ops_since: Mutex<crate::ops_since_http::OpsSinceCache>,
    /// Optional server-imposed limits on the files-beside-the-op-log blob
    /// space (#1007): `/v1/blobs/batch` refuses an oversize blob (413) or
    /// one that would take the store past its quota (507), and
    /// `/v1/ops/batch` refuses a `SetFiles` whose manifest has too many
    /// entries. `None` (the default, single-tenant `lex serve`) is
    /// unlimited. A hosted, multi-tenant embedder such as lex-hub should
    /// set it — the same shape as [`policy_ceiling`](State::policy_ceiling).
    pub blob_limits: Option<BlobLimits>,
    /// `produced_by.tool` names that clients may NOT claim. Defaults to
    /// empty (single-tenant `lex serve` and existing embedders behave
    /// exactly as before). When non-empty, `POST /v1/attestations/batch`
    /// refuses — `403` `ReservedProducer`, whole batch, nothing written —
    /// any attestation whose `produced_by.tool` matches an entry. An entry
    /// is an exact name, or — when it ends in `*` — a prefix
    /// (`"lex-store::review:*"` reserves every `lex-store::review:<who>`
    /// producer, whose suffix is variable). Matching trims and ASCII
    /// case-folds both sides; blank entries are ignored.
    ///
    /// The point is to keep a name that only the server writes (the hub's
    /// own `lex-hub-ci`, see `lex_store::HUB_CI_PRODUCER_TOOL`, and the
    /// `lex-store::review:*` family, see `lex_store::REVIEW_PRODUCER_RESERVATION`)
    /// from being minted by a tenant key holder. Server-internal writers
    /// (`Store::verify_head_and_attest`, `record_review`, …) call the
    /// store directly and are unaffected. A hosted embedder such as
    /// lex-hub should set it; there is deliberately no default name here.
    pub reserved_producers: Vec<String>,
    /// Attestation KINDS that clients may NOT file (#1066). Defaults to
    /// empty (single-tenant `lex serve` and existing embedders behave
    /// exactly as before). When non-empty, `POST /v1/attestations/batch`
    /// refuses — `403` `ReservedKind`, whole batch, nothing written — any
    /// attestation whose kind matches an entry.
    ///
    /// This is the counterpart of [`reserved_producers`](State::reserved_producers)
    /// for readers that key on the KIND rather than on who produced it:
    /// `Store::latest_review_verdict` (the review inbox, `promote`'s
    /// "standing Reject") takes the latest `Review` on a stage whatever its
    /// `produced_by.tool`, so reserving the `lex-store::review:*` producer
    /// family alone does not protect verdicts — a client files a `Review`
    /// under `evil-tool`. An embedder that stamps reviewer identity
    /// server-side reserves [`REVIEW_KIND`] as well.
    ///
    /// An entry is the serde tag of `lex_vcs::AttestationKind`
    /// (`"review"`, `"type_check"`, …; see the `*_KIND` consts). Matching
    /// is against the kind of the PARSED attestation — exactly what the
    /// store would persist — so no body shape can store a reserved kind
    /// without being judged as it (a body with duplicate `"kind"` keys does
    /// not parse at all: `400`).
    /// Matching trims and ASCII case-folds both sides, and additionally
    /// ignores `_`, so `"Review"`, `" REVIEW "`, `"TypeCheck"` and
    /// `"type_check"` all mean what they say; an entry ending in `*` is a
    /// prefix (as for producers); blank entries are ignored.
    ///
    /// Server-internal writers (`Store::verify_head_and_attest`,
    /// `record_review`, `POST /v1/review/verdict`, …) call the store
    /// directly and are unaffected. Reserving a kind is not authenticity
    /// (a signature is): reserving `TypeCheck` should wait for
    /// signature-checked gates.
    pub reserved_kinds: Vec<String>,
}

/// Serde tag of `AttestationKind::Review` — the value an embedder passes to
/// [`State::with_reserved_kinds`] to keep verdicts server-stamped (#1066).
/// Pair it with `lex_store::REVIEW_PRODUCER_RESERVATION`.
pub const REVIEW_KIND: &str = "review";

/// Serde tag of `AttestationKind::TypeCheck` (#1066). Reserving it should
/// wait for signature-checked gates; exported so the name is not retyped.
pub const TYPE_CHECK_KIND: &str = "type_check";

/// Limits on one store's blob space (#1007). See [`State::blob_limits`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BlobLimits {
    /// Largest single blob accepted, in decoded bytes.
    pub max_blob_bytes: u64,
    /// Total decoded bytes the store's blob space may hold.
    pub store_quota_bytes: u64,
    /// Most entries a `SetFiles` manifest may name.
    pub max_manifest_entries: usize,
}

/// Capabilities this server advertises on `/v1/health` (#1007). A client
/// states the ones it speaks in the `X-Lex-Caps` request header
/// (comma-separated).
pub const CAPS: &[&str] = &[CAP_FILES_V1, CAP_INTENT_ORIGIN_V1];

/// The server stores and serves `SetFiles` ops and their blobs (#1007).
pub const CAP_FILES_V1: &str = "files-v1";

/// The server stores and serves `Intent.origin` (#892) — the external-VCS
/// provenance of an imported intent. A server without it deserializes an
/// origin-bearing intent, silently drops the unknown field and re-stores an
/// intent whose bytes no longer match its id, so `lex op push` refuses to
/// send one to a hub that doesn't advertise this.
pub const CAP_INTENT_ORIGIN_V1: &str = "intent-origin-v1";

/// Whether an `X-Lex-Caps` header value names `cap`.
pub(crate) fn has_cap(header: Option<&str>, cap: &str) -> bool {
    header.is_some_and(|h| h.split(',').any(|c| c.trim().eq_ignore_ascii_case(cap)))
}

/// Server-side wrapper around [`MergeSession`] carrying the
/// branch names that started the merge. The lex-vcs session
/// itself only tracks `OpId` heads; commit needs the dst branch
/// name to advance the right head, and the src branch name is
/// kept for round-trip auditability ("which branch did we merge
/// from?").
pub struct ApiMergeSession {
    pub inner: MergeSession,
    pub src_branch: String,
    pub dst_branch: String,
}

/// The entry of `entries` that `claimed` matches: trim + ASCII case-fold
/// (`fold`) both sides; an entry ending in `*` is a prefix; blank entries
/// are ignored. Shared by `reserved_producers` and `reserved_kinds`.
fn find_reserved<'a>(entries: &'a [String], claimed: &str, fold: fn(&str) -> String) -> Option<&'a str> {
    let claimed = fold(claimed);
    entries.iter().map(String::as_str).find(|entry| {
        let entry = fold(entry);
        match entry.strip_suffix('*') {
            Some(prefix) => !prefix.is_empty() && claimed.starts_with(prefix),
            None => !entry.is_empty() && claimed == entry,
        }
    })
}

fn fold_name(s: &str) -> String {
    s.trim().to_ascii_lowercase()
}

/// Like [`fold_name`], and `_` is dropped so `TypeCheck` and `type_check`
/// (the serde tag) are the same kind.
fn fold_kind(s: &str) -> String {
    s.trim().chars().filter(|c| *c != '_').map(|c| c.to_ascii_lowercase()).collect()
}

/// The serde tag of `kind` — the exact `"kind"` string the store persists.
fn attestation_kind_tag(kind: &lex_vcs::AttestationKind) -> String {
    serde_json::to_value(kind)
        .ok()
        .and_then(|v| v.get("kind").and_then(|t| t.as_str().map(str::to_string)))
        .unwrap_or_default()
}

impl State {
    pub fn open(root: PathBuf) -> anyhow::Result<Self> {
        Self::open_with_ceiling(root, None)
    }

    /// Like [`State::open`] but installs a [`policy_ceiling`](State::policy_ceiling)
    /// that `/v1/run` and `/v1/replay` clamp the caller's request
    /// policy against. Embedders exposing this API to untrusted
    /// callers must use this constructor (or set the field directly).
    pub fn open_with_ceiling(
        root: PathBuf,
        policy_ceiling: Option<Policy>,
    ) -> anyhow::Result<Self> {
        Ok(Self {
            store: Mutex::new(Store::open(&root)?),
            root,
            sessions: Mutex::new(HashMap::new()),
            policy_ceiling,
            ops_since: Mutex::new(Default::default()),
            blob_limits: None,
            reserved_producers: Vec::new(),
            reserved_kinds: Vec::new(),
        })
    }

    /// Install [`blob_limits`](State::blob_limits) (#1007).
    pub fn with_blob_limits(mut self, limits: Option<BlobLimits>) -> Self {
        self.blob_limits = limits;
        self
    }

    /// Install [`reserved_producers`](State::reserved_producers): the
    /// `produced_by.tool` names clients may not claim through the
    /// attestation-writing HTTP endpoints.
    pub fn with_reserved_producers(mut self, tools: Vec<String>) -> Self {
        self.reserved_producers = tools;
        self
    }

    /// Install [`reserved_kinds`](State::reserved_kinds): the attestation
    /// kinds clients may not file through `POST /v1/attestations/batch`.
    pub fn with_reserved_kinds(mut self, kinds: Vec<String>) -> Self {
        self.reserved_kinds = kinds;
        self
    }

    /// The reserved producer name `att` claims, if any.
    fn reserved_producer_claimed(&self, att: &lex_vcs::Attestation) -> Option<&str> {
        find_reserved(&self.reserved_producers, &att.produced_by.tool, fold_name)
    }

    /// The reserved kind entry `att`'s kind matches, if any, with the
    /// kind's serde tag. Judges the PARSED kind (what the store persists).
    fn reserved_kind_claimed(&self, att: &lex_vcs::Attestation) -> Option<(&str, String)> {
        if self.reserved_kinds.is_empty() {
            return None;
        }
        let tag = attestation_kind_tag(&att.kind);
        find_reserved(&self.reserved_kinds, &tag, fold_kind).map(|entry| (entry, tag))
    }

    /// Construct a per-tenant `State` by prefixing `store_root` with the
    /// tenant id. Single-tenant `lex serve` is unaffected — it calls
    /// `State::open` directly.
    ///
    /// `tenant_id` is restricted to `[A-Za-z0-9_-]{1,64}`: anything else
    /// (path separators, `..`, NUL, absolute paths, dotfiles, empty
    /// string) is rejected before touching the filesystem. Without this
    /// `PathBuf::join("/etc")` would silently replace `store_root`, and
    /// `PathBuf::join("../foo")` would escape the tenant root.
    pub fn new_with_tenant(tenant_id: &str, store_root: PathBuf) -> anyhow::Result<Self> {
        validate_tenant_id(tenant_id)?;
        Self::open(store_root.join(tenant_id))
    }

    /// Multi-tenant constructor that also installs a policy ceiling
    /// for `/v1/run` / `/v1/replay`. The path-traversal guard from
    /// [`new_with_tenant`](State::new_with_tenant) and the effect
    /// ceiling are the two halves a hosted gateway needs.
    pub fn new_with_tenant_and_ceiling(
        tenant_id: &str,
        store_root: PathBuf,
        policy_ceiling: Option<Policy>,
    ) -> anyhow::Result<Self> {
        validate_tenant_id(tenant_id)?;
        Self::open_with_ceiling(store_root.join(tenant_id), policy_ceiling)
    }
}

/// Clamp a caller-supplied [`Policy`] to a server-imposed `ceiling`
/// so it can only *narrow* the granted capabilities, never widen
/// them. Used by [`run_handler`] when [`State::policy_ceiling`] is
/// set — i.e. when an embedder exposes `/v1/run` to untrusted
/// callers and must not let the request body grant itself `[proc]`,
/// arbitrary `[fs_*]` paths, or unrestricted `[net]`.
///
/// - **Effects**: set-intersection of request and ceiling. The
///   caller may drop effects but never add one the ceiling withheld.
/// - **Scopes** (fs paths, proc binaries, net hosts): taken from the
///   ceiling outright. The caller cannot widen them, and — because an
///   empty scope list means "any" in the runtime — we must not let a
///   caller's empty list collapse the ceiling's restriction back to a
///   wildcard.
/// - **Budget**: the more restrictive (smaller) of the two.
fn clamp_policy(requested: Policy, ceiling: &Policy) -> Policy {
    let allow_effects: BTreeSet<String> = requested
        .allow_effects
        .intersection(&ceiling.allow_effects)
        .cloned()
        .collect();
    let budget = match (requested.budget, ceiling.budget) {
        (Some(r), Some(c)) => Some(r.min(c)),
        (None, Some(c)) => Some(c),
        (Some(r), None) => Some(r),
        (None, None) => None,
    };
    Policy {
        allow_effects,
        allow_fs_read: ceiling.allow_fs_read.clone(),
        allow_fs_write: ceiling.allow_fs_write.clone(),
        allow_net_host: ceiling.allow_net_host.clone(),
        allow_proc: ceiling.allow_proc.clone(),
        allow_approval: ceiling.allow_approval.clone(),
        budget,
    }
}

fn validate_tenant_id(tenant_id: &str) -> anyhow::Result<()> {
    if tenant_id.is_empty() {
        anyhow::bail!("tenant_id must not be empty");
    }
    if tenant_id.len() > 64 {
        anyhow::bail!("tenant_id must be at most 64 bytes");
    }
    if !tenant_id
        .bytes()
        .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-')
    {
        anyhow::bail!(
            "tenant_id {tenant_id:?} contains characters outside [A-Za-z0-9_-]"
        );
    }
    Ok(())
}

#[derive(Debug, Serialize, Deserialize)]
struct ErrorEnvelope {
    error: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    detail: Option<serde_json::Value>,
}

pub(crate) fn json_response(status: u16, body: &serde_json::Value) -> Response<std::io::Cursor<Vec<u8>>> {
    let bytes = serde_json::to_vec(body).unwrap_or_else(|_| b"{}".to_vec());
    Response::from_data(bytes)
        .with_status_code(status)
        .with_header(Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..]).unwrap())
}

pub(crate) fn error_response(status: u16, msg: impl Into<String>) -> Response<std::io::Cursor<Vec<u8>>> {
    json_response(status, &serde_json::to_value(ErrorEnvelope {
        error: msg.into(), detail: None,
    }).unwrap())
}

pub(crate) fn error_with_detail(status: u16, msg: impl Into<String>, detail: serde_json::Value)
    -> Response<std::io::Cursor<Vec<u8>>>
{
    json_response(status, &serde_json::to_value(ErrorEnvelope {
        error: msg.into(), detail: Some(detail),
    }).unwrap())
}

/// The actionable next step for a refused unsatisfiable pair (#992). Shared
/// with the `lex op push` client, which prints it verbatim.
pub const UNSATISFIABLE_PAIR_HINT: &str =
    "republish from source to retire the stranded entry (#995)";

/// `StoreError::UnsatisfiablePair` → **422**, never 500 (#992). The request
/// asked to move a head onto a `(sig, stage)` pair no store can hold — a
/// problem with the client's data, not the server. The body names the pair,
/// the sig that actually owns the stage, and what to do about it:
///
/// ```json
/// { "error": "UnsatisfiablePair",
///   "detail": { "sig_id": "...", "stage_id": "...", "filed_under": "...",
///               "hint": "republish from source to retire the stranded entry (#995)" } }
/// ```
///
/// `None` for any other error, so callers fall through to their own mapping.
pub(crate) fn unsatisfiable_pair_response(err: &lex_store::StoreError)
    -> Option<Response<std::io::Cursor<Vec<u8>>>>
{
    let lex_store::StoreError::UnsatisfiablePair { sig_id, stage_id, filed_under } = err else {
        return None;
    };
    Some(error_with_detail(422, "UnsatisfiablePair", serde_json::json!({
        "sig_id": sig_id,
        "stage_id": stage_id,
        "filed_under": filed_under,
        "message": err.to_string(),
        "hint": UNSATISFIABLE_PAIR_HINT,
    })))
}

/// Map a `StoreError` from a write path (`apply_operation` /
/// `apply_operation_checked`) to an HTTP response. The only special
/// case today is `Contention` (#262 multi-writer CAS retries
/// exhausted), which maps to 503 with a `Retry-After` header so
/// clients back off rather than hammering the same branch tip.
pub(crate) fn write_error_response(prefix: &str, err: lex_store::StoreError)
    -> Response<std::io::Cursor<Vec<u8>>>
{
    if let Some(resp) = unsatisfiable_pair_response(&err) {
        return resp;
    }
    if let lex_store::StoreError::Contention { branch, attempts } = &err {
        let body = serde_json::to_vec(&ErrorEnvelope {
            error: format!("{prefix}: branch '{branch}' is contended (attempts={attempts})"),
            detail: Some(serde_json::json!({
                "kind": "contention",
                "branch": branch,
                "attempts": attempts,
            })),
        }).unwrap_or_else(|_| b"{}".to_vec());
        return Response::from_data(body)
            .with_status_code(503)
            .with_header(Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..]).unwrap())
            .with_header(Header::from_bytes(&b"Retry-After"[..], &b"1"[..]).unwrap());
    }
    // #292 slice 3: budget overflow → 503 with `Retry-After: 0`.
    // Unlike Contention (where a retry might land after another
    // writer finishes), there's no point retrying a budget-
    // exceeded op — the caller needs to raise the cap, switch
    // sessions, or refactor the work. The `Retry-After: 0`
    // signals "don't bother retrying as-is" while still using
    // the canonical "service refused" status code.
    if let lex_store::StoreError::BudgetExceeded { session_id, cap, spent_after } = &err {
        let body = serde_json::to_vec(&ErrorEnvelope {
            error: format!(
                "{prefix}: session `{session_id}` budget exceeded \
                 (spent_after={spent_after}, cap={cap})"
            ),
            detail: Some(serde_json::json!({
                "kind": "budget_exceeded",
                "session_id": session_id,
                "cap": cap,
                "spent_after": spent_after,
            })),
        }).unwrap_or_else(|_| b"{}".to_vec());
        return Response::from_data(body)
            .with_status_code(503)
            .with_header(Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..]).unwrap())
            .with_header(Header::from_bytes(&b"Retry-After"[..], &b"0"[..]).unwrap());
    }
    error_response(500, format!("{prefix}: {err}"))
}

pub fn handle(state: Arc<State>, mut req: Request) -> std::io::Result<()> {
    let method = req.method().clone();
    let url = req.url().to_string();
    let path = url.split('?').next().unwrap_or("").to_string();
    let query = url.split_once('?').map(|(_, q)| q.to_string()).unwrap_or_default();

    // `X-Lex-User` is the v3d session identifier — set by humans
    // operating the web UI through whatever proxy fronts auth, or
    // by AI agents calling the JSON API. We pluck it once here so
    // every handler can take it as a borrowed string.
    let x_lex_user = req.headers().iter()
        .find(|h| h.field.equiv("x-lex-user"))
        .map(|h| h.value.as_str().to_string());
    // `X-Lex-Caps` (#1007): the capabilities the client speaks, so a route
    // can refuse to hand an old client history it would mis-store.
    let x_lex_caps = req.headers().iter()
        .find(|h| h.field.equiv("x-lex-caps"))
        .map(|h| h.value.as_str().to_string());

    // POST /v1/pkg/publish sends a raw tar.gz body — read bytes before routing.
    if matches!(method, Method::Post) && path == "/v1/pkg/publish" {
        let mut body_bytes: Vec<u8> = Vec::new();
        let _ = req.as_reader().read_to_end(&mut body_bytes);
        let resp = pkg_publish_handler(&state, &body_bytes);
        return req.respond(resp);
    }

    let mut body = String::new();
    let _ = req.as_reader().read_to_string(&mut body);

    let resp = route(&state, &method, &path, &query, &body, x_lex_user.as_deref(), x_lex_caps.as_deref());
    req.respond(resp)
}

/// Auth-gated entry point. Calls `auth(path, headers)` before routing;
/// returns 401 JSON when it returns false. Keeps auth logic out of the
/// product-agnostic core.
pub fn handle_with_auth<F>(state: Arc<State>, req: Request, auth: F) -> std::io::Result<()>
where
    F: FnOnce(&str, &[Header]) -> bool,
{
    let path = req.url().split('?').next().unwrap_or("").to_string();
    if !auth(&path, req.headers()) {
        return req.respond(
            Response::from_data(br#"{"error":"unauthorized"}"#.to_vec())
                .with_status_code(401)
                .with_header(
                    Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..]).unwrap(),
                ),
        );
    }
    handle(state, req)
}

fn route(
    state: &State,
    method: &Method,
    path: &str,
    query: &str,
    body: &str,
    x_lex_user: Option<&str>,
    x_lex_caps: Option<&str>,
) -> Response<std::io::Cursor<Vec<u8>>> {
    match (method, path) {
        // ---- lex-tea v2 (HTML browser) ------------------------
        (Method::Get, "/") => crate::web::activity_handler(state),
        (Method::Get, "/web/branches") => crate::web::branches_handler(state),
        (Method::Get, "/web/trust") => crate::web::trust_handler(state),
        (Method::Get, "/web/attention") => crate::web::attention_handler(state),
        (Method::Get, p) if p.starts_with("/web/branch/") => {
            let name = &p["/web/branch/".len()..];
            crate::web::branch_handler(state, name)
        }
        (Method::Get, p) if p.starts_with("/web/stage/") => {
            let id = &p["/web/stage/".len()..];
            crate::web::stage_html_handler(state, id)
        }
        // lex-tea v3 human-triage actions (#172). HTML forms post
        // to /web/stage/<id>/{pin,defer,block,unblock} with a
        // `reason` body. All four share one handler; the verb in
        // the path picks the AttestationKind.
        (Method::Post, p) if p.starts_with("/web/stage/") && (
            p.ends_with("/pin") || p.ends_with("/defer")
            || p.ends_with("/block") || p.ends_with("/unblock")
        ) => {
            let prefix_len = "/web/stage/".len();
            let last_slash = p.rfind('/').unwrap_or(p.len());
            let id = &p[prefix_len..last_slash];
            let verb = &p[last_slash + 1..];
            let decision = match verb {
                "pin"     => crate::web::WebStageDecision::Pin,
                "defer"   => crate::web::WebStageDecision::Defer,
                "block"   => crate::web::WebStageDecision::Block,
                "unblock" => crate::web::WebStageDecision::Unblock,
                _ => unreachable!("matched in outer guard"),
            };
            crate::web::stage_decision_handler(state, id, body, decision, x_lex_user)
        }
        // ---- JSON API -----------------------------------------
        (Method::Get, "/v1/health") => json_response(200, &serde_json::json!({"ok": true, "caps": CAPS})),
        (Method::Post, "/v1/parse") => parse_handler(body),
        (Method::Post, "/v1/check") => check_handler(body),
        (Method::Post, "/v1/publish") => publish_handler(state, body),
        (Method::Post, "/v1/patch") => patch_handler(state, body),
        (Method::Post, "/v1/transform") => crate::transform_http::transform_handler(state, body),
        (Method::Get, p) if p.starts_with("/v1/stage/") => {
            let suffix = &p["/v1/stage/".len()..];
            // Match `/v1/stage/<id>/attestations` first so a literal
            // stage_id of "attestations" can't be misrouted.
            if let Some(id) = suffix.strip_suffix("/attestations") {
                stage_attestations_handler(state, id)
            } else {
                stage_handler(state, suffix)
            }
        }
        (Method::Post, "/v1/run") => run_handler(state, body, false),
        (Method::Post, "/v1/replay") => run_handler(state, body, true),
        (Method::Get, p) if p.starts_with("/v1/trace/") => {
            let id = &p["/v1/trace/".len()..];
            trace_handler(state, id)
        }
        (Method::Get, "/v1/diff") => diff_handler(state, query),
        (Method::Post, "/v1/merge/start") => merge_start_handler(state, body),
        (Method::Post, p) if p.starts_with("/v1/merge/") && p.ends_with("/resolve") => {
            let id = &p["/v1/merge/".len()..p.len() - "/resolve".len()];
            merge_resolve_handler(state, id, body)
        }
        (Method::Post, p) if p.starts_with("/v1/merge/") && p.ends_with("/commit") => {
            let id = &p["/v1/merge/".len()..p.len() - "/commit".len()];
            merge_commit_handler(state, id)
        }
        // ---- #242: append-only sync of op log + attestation log
        (Method::Post, "/v1/ops/batch") => ops_batch_handler(state, body),
        (Method::Post, "/v1/attestations/batch") => attestations_batch_handler(state, body),
        // Content half of push/pull: the stage (code) and intent blobs the
        // op records reference. `batch` receives, `fetch` returns by id.
        (Method::Post, "/v1/stages/batch") => crate::sync_http::stages_batch_handler(state, body),
        (Method::Post, "/v1/stages/fetch") => crate::sync_http::stages_fetch_handler(state, body),
        (Method::Post, "/v1/stages/missing") => crate::sync_http::stages_missing_handler(state, body),
        (Method::Post, "/v1/intents/batch") => crate::sync_http::intents_batch_handler(state, body),
        (Method::Post, "/v1/intents/fetch") => crate::sync_http::intents_fetch_handler(state, body),
        // #930 P2b-1: committed lockfiles travel with the package so the
        // write-time gate can resolve a head's pinned dependencies.
        (Method::Post, "/v1/locks/batch") => crate::sync_http::locks_batch_handler(state, body),
        (Method::Post, "/v1/locks/fetch") => crate::sync_http::locks_fetch_handler(state, body),
        // #1007: files beside the op-log — content-addressed blobs a
        // `SetFiles` manifest names. Pushed before the ops that need them.
        (Method::Post, "/v1/blobs/missing") => crate::sync_http::blobs_missing_handler(state, body),
        (Method::Post, "/v1/blobs/batch") => crate::sync_http::blobs_batch_handler(state, body),
        (Method::Post, "/v1/blobs/fetch") => crate::sync_http::blobs_fetch_handler(state, body),
        // #949 phase 1: typed issues travel with the package (content-
        // addressed, like intents) so a pulled op-log carries its work items.
        (Method::Post, "/v1/issues/batch") => crate::sync_http::issues_batch_handler(state, body),
        (Method::Post, "/v1/issues/fetch") => crate::sync_http::issues_fetch_handler(state, body),
        (Method::Get, "/v1/issues/list") => crate::sync_http::issues_list_handler(state),
        // #949 phase 3: derived issue/project state, computed from the log —
        // a board is a view, nobody drags cards. The `/list` literal above
        // wins over the `/v1/issues/<id>` prefix arm below.
        (Method::Get, "/v1/issues") => crate::issues_http::issues_state_handler(state),
        (Method::Get, "/v1/projects") => crate::issues_http::projects_handler(state),
        (Method::Get, p) if p.starts_with("/v1/issues/") => {
            crate::issues_http::issue_detail_handler(state, &p["/v1/issues/".len()..])
        }
        // ---- #839 follow-up: branch management over HTTP so a remote
        // client can create/switch branches (and thus drive the merge
        // gates end to end), not just probe heads.
        (Method::Get, "/v1/review/inbox") => crate::review_http::review_inbox_handler(state, query),
        (Method::Post, "/v1/review/verdict") => crate::review_http::review_verdict_handler(state, body),
        (Method::Get, "/v1/branches") => crate::branches_http::branches_list_handler(state),
        (Method::Post, "/v1/branches") => crate::branches_http::branch_create_handler(state, body),
        (Method::Post, p) if p.starts_with("/v1/branches/") && p.ends_with("/checkout") => {
            let name = &p["/v1/branches/".len()..p.len() - "/checkout".len()];
            crate::branches_http::branch_checkout_handler(state, name)
        }
        // Branch head: GET probes it (for `op push`'s delta), POST advances
        // it (the ref half of push, fast-forward-only). Both in branches_http.
        (Method::Get, p) if p.starts_with("/v1/branches/") && p.ends_with("/head") => {
            let name = &p["/v1/branches/".len()..p.len() - "/head".len()];
            crate::branches_http::branch_head_handler(state, name)
        }
        (Method::Post, p) if p.starts_with("/v1/branches/") && p.ends_with("/head") => {
            let name = &p["/v1/branches/".len()..p.len() - "/head".len()];
            crate::branches_http::branch_advance_head_handler(state, name, body)
        }
        // ---- #260: append-only fetch (inverse of #242 push)
        // Body is a JSON array of OperationRecords reachable from
        // `branch.head_op` but not from `after`, oldest-first.
        (Method::Get, "/v1/ops/since") => crate::ops_since_http::ops_since_handler(state, query, x_lex_caps),
        (Method::Get, "/v1/attestations/since") => attestations_since_handler(state, query),
        // ---- #4: package concept ----------------------------------
        // POST /v1/pkg/publish is handled in handle() before route()
        // (binary body), so it doesn't appear here.
        (Method::Get, "/v1/pkg") => pkg_list_handler(state),
        // Owner-only visibility toggle (authed via the front door). Must
        // precede the generic `/v1/pkg/{name}` arms; it's a PUT, so it
        // can't collide with the GET/DELETE arms regardless.
        (Method::Put, p) if p.starts_with("/v1/pkg/") && p.ends_with("/visibility") => {
            let name = &p["/v1/pkg/".len()..p.len() - "/visibility".len()];
            pkg_set_visibility_handler(state, name, body)
        }
        // Cut an immutable versioned release of an op-log-hosted package
        // (#893). POST, before the generic /v1/pkg/{name} arms.
        (Method::Post, p) if p.starts_with("/v1/pkg/") && p.ends_with("/release") => {
            let name = &p["/v1/pkg/".len()..p.len() - "/release".len()];
            pkg_release_handler(state, name, body)
        }
        (Method::Get, p) if p.starts_with("/v1/pkg/") && p.ends_with("/head") => {
            let name = &p["/v1/pkg/".len()..p.len() - "/head".len()];
            pkg_head_handler(state, name)
        }
        (Method::Get, p) if p.starts_with("/v1/pkg/") && p.ends_with("/versions") => {
            let name = &p["/v1/pkg/".len()..p.len() - "/versions".len()];
            pkg_versions_handler(state, name)
        }
        (Method::Get, p) if p.starts_with("/v1/pkg/") && p.ends_with("/api-diff") => {
            let name = &p["/v1/pkg/".len()..p.len() - "/api-diff".len()];
            pkg_api_diff_handler(state, name, query)
        }
        // /v1/pkg/{name}/{version}/archive — must match before the generic /{name}/{version}
        (Method::Get, p) if p.starts_with("/v1/pkg/") && p.ends_with("/archive") => {
            let inner = &p["/v1/pkg/".len()..p.len() - "/archive".len()];
            // inner = "{name}/{version}"
            if let Some((name, version)) = inner.split_once('/') {
                pkg_archive_handler(state, name, version)
            } else {
                error_response(400, "expected /v1/pkg/{name}/{version}/archive")
            }
        }
        // /v1/pkg/{name}/{version}
        (Method::Get, p) if p.starts_with("/v1/pkg/") && p["/v1/pkg/".len()..].contains('/') => {
            let inner = &p["/v1/pkg/".len()..];
            if let Some((name, version)) = inner.split_once('/') {
                pkg_get_version_handler(state, name, version)
            } else {
                error_response(400, "expected /v1/pkg/{name}/{version}")
            }
        }
        (Method::Get, p) if p.starts_with("/v1/pkg/") => {
            let name = &p["/v1/pkg/".len()..];
            pkg_get_handler(state, name)
        }
        (Method::Delete, p) if p.starts_with("/v1/pkg/") => {
            let name = &p["/v1/pkg/".len()..];
            pkg_delete_handler(state, name)
        }
        _ => error_response(404, format!("unknown route: {method:?} {path}")),
    }
}

#[derive(Deserialize)]
struct ParseReq { source: String }

fn parse_handler(body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let req: ParseReq = match serde_json::from_str(body) {
        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
    };
    match load_program_from_str(&req.source) {
        Ok(prog) => {
            let stages = canonicalize_program(&prog);
            json_response(200, &serde_json::to_value(&stages).unwrap())
        }
        Err(e) => error_response(400, format!("syntax error: {e}")),
    }
}

pub(crate) fn check_handler(body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let req: ParseReq = match serde_json::from_str(body) {
        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
    };
    let prog = match load_program_from_str(&req.source) {
        Ok(p) => p, Err(e) => return error_response(400, format!("syntax error: {e}")),
    };
    let stages = canonicalize_program(&prog);
    match lex_types::check_program(&stages) {
        Ok(_) => json_response(200, &serde_json::json!({"ok": true})),
        Err(errs) => json_response(422, &serde_json::to_value(&errs).unwrap()),
    }
}

#[derive(Deserialize)]
struct PublishReq { source: String, #[serde(default)] activate: bool }

pub(crate) fn publish_handler(state: &State, body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let req: PublishReq = match serde_json::from_str(body) {
        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
    };
    let prog = match load_program_from_str(&req.source) {
        Ok(p) => p, Err(e) => return error_response(400, format!("syntax error: {e}")),
    };
    // #168: rewrite stdlib parse calls to parse_strict so the
    // bytecode emitted from these stages enforces required-field
    // checks at runtime.
    let mut stages = canonicalize_program(&prog);
    if let Err(errs) = lex_types::check_and_rewrite_program(&mut stages) {
        return error_with_detail(422, "type errors", serde_json::to_value(&errs).unwrap());
    }
    // #835 Tier 1: behavioral example gate. check_and_rewrite_program
    // only type-checks `examples {}`; run them and refuse the publish
    // if any declared example evaluates to the wrong value. Non-breaking:
    // functions without examples (and effectful ones, which can't have
    // them) produce no cases.
    let example_errors = lex_runtime::evaluate_examples(&stages);
    if !example_errors.is_empty() {
        return error_with_detail(422, "example mismatch",
            serde_json::to_value(&example_errors).unwrap_or_default());
    }

    let store = state.store.lock().unwrap();
    let branch = store.current_branch();

    // Compute diff between what's already on the branch and the new program.
    let old_head = match store.branch_head(&branch) {
        Ok(h) => h,
        Err(e) => return error_response(500, format!("branch_head: {e}")),
    };
    // Fns + types (#895) on both sides. Old side is the branch head, read
    // in one pass through the SigId the head names each stage by (#971):
    // `get_ast` per entry re-read the whole stage index once per live
    // declaration, and — StageIds being name-independent — resolved two
    // functions differing only in name to one of the two, so the other was
    // re-reported as an Add on every republish (#826).
    let old_pairs: Vec<(String, String)> =
        old_head.iter().map(|(sig, stg)| (sig.clone(), stg.clone())).collect();
    let old_head_stages: Vec<lex_ast::Stage> =
        store.get_asts_for_sigs_bulk(&old_pairs).into_iter().filter_map(Result::ok).collect();
    let old_fns = stage_fns(&old_head_stages);
    let new_fns = stage_fns(&stages);
    let old_types = stage_types(&old_head_stages);
    let new_types = stage_types(&stages);
    let report =
        lex_vcs::compute_diff_with_types(&old_fns, &new_fns, &old_types, &new_types, false);

    // Build new imports map from any Import stages in the source.
    let mut new_imports: lex_vcs::ImportMap = lex_vcs::ImportMap::new();
    {
        let entry = new_imports.entry("<source>".into()).or_default();
        for s in &stages {
            if let lex_ast::Stage::Import(im) = s {
                entry.insert(lex_vcs::ImportRef {
                    reference: im.reference.clone(),
                    alias: im.alias.clone(),
                });
            }
        }
    }

    match store.publish_program(&branch, &stages, &report, &new_imports, req.activate) {
        Ok(outcome) => {
            // #835 Tier 1: record the behavioral-examples verdict for each
            // published fn-stage that declares examples. Best-effort — a
            // failure to record must not fail an otherwise-good publish.
            record_examples_for_publish(&store, &stages, &outcome);
            json_response(200, &serde_json::json!({
                "ops": outcome.ops,
                "head_op": outcome.head_op,
            }))
        }
        // The store-write gate (#130) also type-checks at the top
        // of `publish_program`. The handler above already pre-checks,
        // so this branch is reached only on a race or a state we
        // didn't see at handler time. Surface the structured
        // envelope (422) instead of a generic 500 — same shape the
        // initial pre-check uses, so a client only has one error
        // contract to handle.
        Err(lex_store::StoreError::TypeError(errs)) => {
            error_with_detail(422, "type errors", serde_json::to_value(&errs).unwrap())
        }
        Err(e) => write_error_response("publish_program", e),
    }
}

#[derive(Deserialize)]
struct PatchReq {
    stage_id: String,
    patch: lex_ast::Patch,
    #[serde(default)] activate: bool,
    /// #837 piece A: the branch to write to. Absent keeps the historical
    /// behaviour (the server's global current branch); a supplied name is
    /// honoured, and an unknown one is a 404.
    #[serde(default)] branch: Option<String>,
    /// #837 piece A: attribute the write. Absent records no intent, as before.
    #[serde(default)] intent: Option<crate::transform_http::IntentSpec>,
}

/// POST /v1/patch — apply a structured edit to a stored stage's
/// canonical AST, type-check the result, and publish a new stage.
///
/// Optional `branch` / `intent` (#837 piece A) name the branch explicitly and
/// attribute the op; without them the request behaves exactly as it always did.
/// For the typed transforms, and a request that *requires* a branch, see
/// `POST /v1/transform`.
fn patch_handler(state: &State, body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let req: PatchReq = match serde_json::from_str(body) {
        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
    };
    let intent = match req.intent.clone() {
        Some(spec) => match spec.into_intent(crate::transform_http::default_http_session) {
            Ok(i) => Some(i),
            Err(e) => return error_response(400, format!("bad request: {e}")),
        },
        None => None,
    };
    let store = state.store.lock().unwrap();
    let explicit_branch = req.branch.is_some();
    if let Some(b) = &req.branch {
        match store.list_branches() {
            Ok(bs) if bs.iter().any(|x| x == b) => {}
            Ok(_) => return error_response(404, format!("unknown branch `{b}`")),
            Err(e) => return error_response(500, format!("list_branches: {e}")),
        }
    }

    // 1. Load.
    let original = match store.get_ast(&req.stage_id) {
        Ok(s) => s, Err(e) => return error_response(404, format!("stage: {e}")),
    };

    // 2. Apply.
    let patched = match lex_ast::apply_patch(&original, &req.patch) {
        Ok(s) => s,
        Err(e) => return error_with_detail(422, "patch failed",
            serde_json::to_value(&e).unwrap_or_default()),
    };

    // 3. No isolated check (#833): the gated apply below type-checks
    // the *composed* program — the branch head with the patched stage
    // swapped in. Stricter where it matters (a body that no longer
    // composes with its callers is refused) and correct where the old
    // isolated check was wrong (a body calling a sibling was rejected
    // as an unknown identifier a one-stage program couldn't see).

    // Routing through the gated apply so /v1/patch participates in the
    // op DAG. We know this op is always a body change on the existing
    // sig (a patch can't add a brand-new fn).
    let branch = req.branch.clone().unwrap_or_else(|| store.current_branch());

    // Find the sig — patched stage's sig must match the original's.
    let sig = match lex_ast::sig_id(&patched) {
        Some(s) => s,
        None => return error_response(500, "patched stage has no sig_id"),
    };

    // Persist before the gate (its RepairHint on rejection is
    // addressed to this stage); activate only once the head moved.
    let new_id = match store.publish(&patched) {
        Ok(id) => id, Err(e) => return error_response(500, format!("publish: {e}")),
    };

    // Determine op kind: ChangeEffectSig if effects differ, ModifyBody otherwise.
    let original_effects: std::collections::BTreeSet<String> = match &original {
        lex_ast::Stage::FnDecl(fd) => fd.effects.iter().map(|e| e.name.clone()).collect(),
        _ => std::collections::BTreeSet::new(),
    };
    let patched_effects: std::collections::BTreeSet<String> = match &patched {
        lex_ast::Stage::FnDecl(fd) => fd.effects.iter().map(|e| e.name.clone()).collect(),
        _ => std::collections::BTreeSet::new(),
    };
    let head_now = match store.get_branch(&branch) {
        Ok(b) => b.and_then(|b| b.head_op),
        Err(e) => return error_response(500, format!("get_branch: {e}")),
    };
    let kind = if original_effects != patched_effects {
        // #247: budget delta is part of the canonical payload now.
        // Patch endpoints don't currently rehydrate the AST to
        // recompute budgets, so leave them None — clients that
        // need budget tracking should publish through the diff
        // pipeline (`lex publish`) where `compute_diff` populates
        // them.
        let from_budget = lex_vcs::operation_budget_from_effects(&original_effects);
        let to_budget = lex_vcs::operation_budget_from_effects(&patched_effects);
        // #992: the patched stage declares the new effects, so it hashes to a
        // different SigId and the store files it under that one. Record it, or
        // the head keeps the old sig pointing at a stage no store holds there.
        let to_sig_id = lex_ast::sig_id(&patched).filter(|s| *s != sig);
        lex_vcs::OperationKind::ChangeEffectSig {
            sig_id: sig.clone(),
            from_stage_id: req.stage_id.clone(),
            to_stage_id: new_id.clone(),
            from_effects: original_effects,
            to_effects: patched_effects,
            from_budget,
            to_budget,
            to_sig_id,
        }
    } else {
        let budget = lex_vcs::operation_budget_from_effects(&original_effects);
        lex_vcs::OperationKind::ModifyBody {
            sig_id: sig.clone(),
            from_stage_id: req.stage_id.clone(),
            to_stage_id: new_id.clone(),
            from_budget: budget,
            to_budget: budget,
            // #992: a patch that touches the signature (types, examples)
            // moves the sig just as an effect change does.
            to_sig_id: lex_ast::sig_id(&patched).filter(|s| *s != sig),
        }
    };
    // #992: derive the transition from the op, never hard-code `Replace`. A
    // sig-moving op must retire the old sig and bind the new one; a `Replace`
    // here left `(old_sig, new_stage)` at the head — a pair no store holds.
    let transition = lex_store::transition_for_kind(&kind);
    let op = lex_vcs::Operation::new(
        kind,
        head_now.into_iter().collect::<Vec<_>>(),
    );
    let op = match &intent {
        Some(i) => op.with_intent(i.intent_id.clone()),
        None => op,
    };
    let op_id = match store.apply_operation_gated_with_intent(&branch, op, transition, intent.as_ref()) {
        Ok(id) => id,
        Err(lex_store::StoreError::TypeError(errs)) => return error_with_detail(
            422, "type errors after patch", serde_json::to_value(&errs).unwrap_or_default()),
        Err(e) => return write_error_response("apply_operation_gated", e),
    };
    if req.activate {
        if let Err(e) = store.activate(&new_id) {
            return error_response(500, format!("activate: {e}"));
        }
    }

    let status = format!("{:?}",
        store.get_status(&new_id).unwrap_or(lex_store::StageStatus::Draft)).to_lowercase();
    let mut resp = serde_json::json!({
        "old_stage_id": req.stage_id,
        "new_stage_id": new_id,
        "sig_id": sig,
        "status": status,
        "op_id": op_id,
    });
    // Only echoed when the caller opted in, so a request without `branch` /
    // `intent` gets the response it always got, byte for byte.
    if explicit_branch {
        resp["branch"] = serde_json::json!(branch);
    }
    if let Some(i) = &intent {
        resp["intent_id"] = serde_json::json!(i.intent_id);
    }
    json_response(200, &resp)
}

pub(crate) fn stage_handler(state: &State, id: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let store = state.store.lock().unwrap();
    let meta = match store.get_metadata(id) {
        Ok(m) => m, Err(e) => return error_response(404, format!("{e}")),
    };
    let ast = match store.get_ast(id) {
        Ok(a) => a, Err(e) => return error_response(404, format!("{e}")),
    };
    let status = format!("{:?}", store.get_status(id).unwrap_or(lex_store::StageStatus::Draft)).to_lowercase();
    json_response(200, &serde_json::json!({
        "metadata": meta,
        "ast": ast,
        "status": status,
    }))
}

/// `GET /v1/stage/<id>/attestations` — every persisted attestation
/// for this stage, newest-first by arrival order. Issue #132's
/// queryable-evidence consumer surface.
///
/// 404s on unknown stage_id (matches `/v1/stage/<id>`'s shape so a
/// caller round-tripping both endpoints sees consistent errors).
/// Empty list (200) is *evidence of absence*: the stage exists but
/// no producer has attested it.
pub(crate) fn stage_attestations_handler(state: &State, id: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let store = state.store.lock().unwrap();
    if let Err(e) = store.get_metadata(id) {
        return error_response(404, format!("{e}"));
    }
    let log = match store.attestation_log() {
        Ok(l) => l,
        Err(e) => return error_response(500, format!("attestation log: {e}")),
    };
    // Newest-first by ARRIVAL order (server-assigned), not by the
    // writer-supplied timestamp; legacy unstamped entries sort below
    // stamped ones. `lex op pull` re-puts these in reverse (oldest
    // first) so a puller's local arrival order matches this store's.
    let mut listing = match log.list_for_stage_by_arrival(&id.to_string()) {
        Ok(v) => v,
        Err(e) => return error_response(500, format!("list_for_stage: {e}")),
    };
    listing.reverse();
    json_response(200, &serde_json::json!({"attestations": listing}))
}

#[derive(Deserialize, Default)]
struct PolicyJson {
    #[serde(default)] allow_effects: Vec<String>,
    #[serde(default)] allow_fs_read: Vec<String>,
    #[serde(default)] allow_fs_write: Vec<String>,
    #[serde(default)] budget: Option<u64>,
}

impl PolicyJson {
    fn into_policy(self) -> Policy {
        Policy {
            allow_effects: self.allow_effects.into_iter().collect::<BTreeSet<_>>(),
            allow_fs_read: self.allow_fs_read.into_iter().map(PathBuf::from).collect(),
            allow_fs_write: self.allow_fs_write.into_iter().map(PathBuf::from).collect(),
            allow_net_host: Vec::new(),
            allow_proc: Vec::new(),
            allow_approval: Vec::new(),
            budget: self.budget,
        }
    }
}

#[derive(Deserialize)]
struct RunReq {
    source: String,
    #[serde(rename = "fn")] func: String,
    #[serde(default)] args: Vec<serde_json::Value>,
    #[serde(default)] policy: PolicyJson,
    #[serde(default)] overrides: IndexMap<String, serde_json::Value>,
}

pub(crate) fn run_handler(state: &State, body: &str, with_overrides: bool) -> Response<std::io::Cursor<Vec<u8>>> {
    let req: RunReq = match serde_json::from_str(body) {
        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
    };
    let prog = match load_program_from_str(&req.source) {
        Ok(p) => p, Err(e) => return error_response(400, format!("syntax error: {e}")),
    };
    let stages = canonicalize_program(&prog);
    if let Err(errs) = lex_types::check_program(&stages) {
        return error_with_detail(422, "type errors", serde_json::to_value(&errs).unwrap());
    }
    let bc = compile_program(&stages);
    let mut policy = req.policy.into_policy();
    // When a server-imposed ceiling is present (multi-tenant
    // embedders like lex-hub), the request policy can only narrow
    // it — never grant itself proc/fs/net beyond what the operator
    // allowed. Single-tenant `lex serve` leaves the ceiling unset
    // and runs the caller's policy verbatim.
    if let Some(ceiling) = &state.policy_ceiling {
        policy = clamp_policy(policy, ceiling);
    }
    if let Err(violations) = check_policy(&bc, &policy) {
        return error_with_detail(403, "policy violation", serde_json::to_value(&violations).unwrap());
    }

    let mut recorder = lex_trace::Recorder::new();
    if with_overrides && !req.overrides.is_empty() {
        recorder = recorder.with_overrides(req.overrides);
    }
    let handle = recorder.handle();
    let handler = DefaultHandler::new(policy);
    let mut vm = Vm::with_handler(&bc, Box::new(handler));
    vm.set_tracer(Box::new(recorder));

    let vargs: Vec<Value> = req.args.iter().map(json_to_value).collect();
    let started = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs();
    let result = vm.call(&req.func, vargs);
    let ended = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs();

    let store = state.store.lock().unwrap();
    let (root_out, root_err, status) = match &result {
        Ok(v) => (Some(value_to_json(v)), None, 200u16),
        Err(e) => (None, Some(format!("{e}")), 200u16),
    };
    let tree = handle.finalize(req.func.clone(), serde_json::Value::Null,
        root_out.clone(), root_err.clone(), started, ended);
    let run_id = match store.save_trace(&tree) {
        Ok(id) => id,
        Err(e) => return error_response(500, format!("save_trace: {e}")),
    };

    let mut body = serde_json::json!({
        "run_id": run_id,
        "output": root_out,
    });
    if let Some(err) = root_err {
        body["error"] = serde_json::Value::String(err);
    }
    json_response(status, &body)
}

fn trace_handler(state: &State, id: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let store = state.store.lock().unwrap();
    match store.load_trace(id) {
        Ok(t) => json_response(200, &serde_json::to_value(&t).unwrap()),
        Err(e) => error_response(404, format!("{e}")),
    }
}

fn diff_handler(state: &State, query: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let mut a = None;
    let mut b = None;
    for kv in query.split('&') {
        if let Some((k, v)) = kv.split_once('=') {
            match k { "a" => a = Some(v.to_string()), "b" => b = Some(v.to_string()), _ => {} }
        }
    }
    let (Some(a), Some(b)) = (a, b) else {
        return error_response(400, "missing a or b query params");
    };
    let store = state.store.lock().unwrap();
    let ta = match store.load_trace(&a) { Ok(t) => t, Err(e) => return error_response(404, format!("a: {e}")) };
    let tb = match store.load_trace(&b) { Ok(t) => t, Err(e) => return error_response(404, format!("b: {e}")) };
    match lex_trace::diff_runs(&ta, &tb) {
        Some(d) => json_response(200, &serde_json::to_value(&d).unwrap()),
        None => json_response(200, &serde_json::json!({"divergence": null})),
    }
}

fn json_to_value(v: &serde_json::Value) -> Value { Value::from_json(v) }

fn value_to_json(v: &Value) -> serde_json::Value { v.to_json() }

#[derive(Deserialize)]
struct MergeStartReq {
    src_branch: String,
    dst_branch: String,
}

/// `POST /v1/merge/start` (#134) — open a stateful merge between two
/// branch heads and return the conflicts the agent needs to
/// resolve. Auto-resolved sigs (one-sided changes, identical
/// changes both sides) are returned for audit but don't block
/// commit.
///
/// Response: `{ merge_id, src_head, dst_head, lca, conflicts,
/// auto_resolved_count }`. The session is held in process memory
/// keyed by `merge_id` for subsequent `resolve` / `commit` calls
/// (next slices).
fn merge_start_handler(state: &State, body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let req: MergeStartReq = match serde_json::from_str(body) {
        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
    };
    let store = state.store.lock().unwrap();
    let src_head = match store.get_branch(&req.src_branch) {
        Ok(Some(b)) => b.head_op,
        Ok(None) => return error_response(404, format!("unknown src branch `{}`", req.src_branch)),
        Err(e) => return error_response(500, format!("src branch read: {e}")),
    };
    let dst_head = match store.get_branch(&req.dst_branch) {
        Ok(Some(b)) => b.head_op,
        Ok(None) => return error_response(404, format!("unknown dst branch `{}`", req.dst_branch)),
        Err(e) => return error_response(500, format!("dst branch read: {e}")),
    };
    let log = match lex_vcs::OpLog::open(store.root()) {
        Ok(l) => l,
        Err(e) => return error_response(500, format!("op log: {e}")),
    };
    // Caller doesn't choose merge_ids — minted server-side from
    // wall clock + a per-process counter avoids leaking session
    // ids' shape into the public surface.
    let merge_id = mint_merge_id();
    let mut session = match MergeSession::start(
        merge_id.clone(),
        &log,
        src_head.as_ref(),
        dst_head.as_ref(),
    ) {
        Ok(s) => s,
        Err(e) => return error_response(500, format!("merge start: {e}")),
    };
    // #1007 PR 7: the files dimension of the merge. `lex-vcs` doesn't
    // know about manifests (see `merge_session.rs`'s module docs), so
    // the store computes the 3-way diff and the session just tracks the
    // conflicts it surfaced — the same layering `MergeResolutionChecker`
    // already uses for sig-conflict type-checking.
    let manifest_outcome = match store.manifest_merge(
        session.lca.as_deref(),
        dst_head.as_deref(),
        src_head.as_deref(),
    ) {
        Ok(o) => o,
        Err(lex_store::StoreError::AmbiguousManifest { op_id }) => {
            return error_with_detail(422, "AmbiguousManifest", serde_json::json!({
                "op_id": op_id,
                "reason": "a merge ancestor has an ambiguous files manifest; \
                           append a set_files op resolving it before merging",
            }));
        }
        Err(e) => return error_response(500, format!("manifest merge: {e}")),
    };
    let (file_conflicts, needs_setfiles): (Vec<lex_vcs::FileConflict>, bool) =
        match manifest_outcome {
            lex_store::ManifestMergeOutcome::NoChange => (Vec::new(), false),
            lex_store::ManifestMergeOutcome::Needed { conflicts, .. } => (conflicts, true),
        };
    session.attach_file_conflicts(file_conflicts, needs_setfiles);

    let conflicts: Vec<&lex_vcs::ConflictRecord> = session.remaining_conflicts();
    let auto_resolved_count = session.auto_resolved.len();
    let remaining_file_conflicts: Vec<&lex_vcs::FileConflict> = session.remaining_file_conflicts();
    let body = serde_json::json!({
        "merge_id": merge_id,
        "src_head": session.src_head,
        "dst_head": session.dst_head,
        "lca":      session.lca,
        "conflicts": conflicts,
        "auto_resolved_count": auto_resolved_count,
        "file_conflicts": remaining_file_conflicts,
        "needs_setfiles": session.needs_setfiles(),
    });
    drop(conflicts);
    drop(remaining_file_conflicts);
    drop(store);
    let wrapped = ApiMergeSession {
        inner: session,
        src_branch: req.src_branch,
        dst_branch: req.dst_branch,
    };
    state.sessions.lock().unwrap().insert(merge_id, wrapped);
    json_response(200, &body)
}

#[derive(Deserialize)]
struct MergeResolveReq {
    /// Each entry is `(conflict_id, resolution)`. The resolution is
    /// the same shape as `lex_vcs::Resolution`'s tagged JSON form
    /// — `{"kind":"take_ours"}`, `{"kind":"take_theirs"}`,
    /// `{"kind":"defer"}`, or `{"kind":"custom","op":{...}}`.
    #[serde(default)]
    resolutions: Vec<MergeResolveEntry>,
    /// File conflicts (#1007 PR 7), keyed by path instead of sig id.
    /// Resolution shape is `lex_vcs::FileResolution`'s tagged JSON form
    /// — `{"kind":"take_ours"}`, `{"kind":"take_theirs"}`, or
    /// `{"kind":"defer"}` (no `custom`; see the module docs on
    /// `merge_session::FileResolution` for why).
    #[serde(default)]
    file_resolutions: Vec<MergeFileResolveEntry>,
}

#[derive(Deserialize)]
struct MergeResolveEntry {
    conflict_id: String,
    resolution: lex_vcs::Resolution,
}

#[derive(Deserialize)]
struct MergeFileResolveEntry {
    path: lex_vcs::FilePath,
    resolution: lex_vcs::FileResolution,
}

/// `POST /v1/merge/<id>/resolve` (#134) — submit batched
/// resolutions against the conflicts surfaced by `merge/start`.
/// Returns one verdict per input: accepted (recorded against the
/// session) or rejected (with structured reason). The session
/// stays alive across calls so an agent can iterate.
///
/// Errors:
/// - 404 if `merge_id` doesn't refer to a live session (a typo
///   or a session GC'd by a server restart).
/// - 400 on malformed body.
fn merge_resolve_handler(
    state: &State,
    merge_id: &str,
    body: &str,
) -> Response<std::io::Cursor<Vec<u8>>> {
    let req: MergeResolveReq = match serde_json::from_str(body) {
        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
    };
    let mut sessions = state.sessions.lock().unwrap();
    let Some(wrapped) = sessions.get_mut(merge_id) else {
        return error_response(404, format!("unknown merge_id `{merge_id}`"));
    };
    let pairs: Vec<(String, lex_vcs::Resolution)> = req.resolutions.into_iter()
        .map(|e| (e.conflict_id, e.resolution))
        .collect();
    // #834: type-check each resolution against dst's head at submission
    // time (not only at commit) so the agent's "submit N, see which
    // broke, retry" loop gets its feedback here. The store composes the
    // projected program; the session owns the merge→delta semantics.
    let store = state.store.lock().unwrap();
    let checker = lex_store::MergeResolutionChecker::new(&store, wrapped.dst_branch.clone());
    let verdicts = wrapped.inner.resolve_checked(pairs, &checker);
    drop(store);
    let file_pairs: Vec<(lex_vcs::FilePath, lex_vcs::FileResolution)> = req.file_resolutions
        .into_iter()
        .map(|e| (e.path, e.resolution))
        .collect();
    let file_verdicts = wrapped.inner.resolve_files(file_pairs);
    let remaining: Vec<&lex_vcs::ConflictRecord> = wrapped.inner.remaining_conflicts();
    let remaining_files: Vec<&lex_vcs::FileConflict> = wrapped.inner.remaining_file_conflicts();
    let body = serde_json::json!({
        "verdicts": verdicts,
        "remaining_conflicts": remaining,
        "file_verdicts": file_verdicts,
        "remaining_file_conflicts": remaining_files,
    });
    json_response(200, &body)
}

/// `POST /v1/merge/<id>/commit` (#134) — finalize a merge
/// session. Builds a `Merge` op from the auto-resolved sigs +
/// the conflict resolutions, applies it to the dst branch, and
/// returns the new head op id. The session is dropped on
/// success; the caller would re-run `merge/start` to land
/// further changes.
///
/// Errors:
/// - 404: unknown `merge_id`.
/// - 422: conflicts remaining (pass `Defer` or just don't
///   resolve a conflict and you land here). Body carries the
///   list so the caller knows which still need attention.
/// - 422: a `Custom` resolution was used. The data layer
///   supports them but landing them via HTTP needs an extra
///   pass to apply the custom op against the dst branch
///   first; deferred to a follow-up slice. Use TakeOurs /
///   TakeTheirs for now.
/// - 409: dependency conflict (#977) — the branches pin the same
///   package at different versions, so no merged lock exists. Body
///   `detail` names the package and both versions; nothing is written.
/// - 500: filesystem error while landing the merge op.
fn merge_commit_handler(
    state: &State,
    merge_id: &str,
) -> Response<std::io::Cursor<Vec<u8>>> {
    use std::collections::BTreeMap;
    let wrapped = match state.sessions.lock().unwrap().remove(merge_id) {
        Some(w) => w,
        None => return error_response(404, format!("unknown merge_id `{merge_id}`")),
    };
    let dst_branch = wrapped.dst_branch.clone();
    let src_head = wrapped.inner.src_head.clone();
    let dst_head = wrapped.inner.dst_head.clone();
    let lca = wrapped.inner.lca.clone();
    let auto_resolved = wrapped.inner.auto_resolved.clone();

    // Conflict resolutions (sig conflicts, then file conflicts — #1007 PR 7).
    let commit_out = match wrapped.inner.commit() {
        Ok(r) => r,
        Err(lex_vcs::CommitError::ConflictsRemaining(ids)) => {
            // Re-insert isn't possible since we removed above; the
            // caller will need to re-start. That's acceptable: a
            // commit-with-unresolved-conflicts is operator error.
            return error_with_detail(
                422,
                "conflicts remaining",
                serde_json::json!({"unresolved": ids}),
            );
        }
        Err(lex_vcs::CommitError::FileConflictsRemaining(paths)) => {
            return error_with_detail(
                422,
                "file conflicts remaining",
                serde_json::json!({"unresolved_files": paths}),
            );
        }
    };

    // Translate auto-resolved + resolutions into the StageTransition::Merge
    // entries map. Every sig the merge decided goes in — including the ones
    // dst already has (`take_ours`, dst-only changes) — so the merged head
    // does not depend on the order a replay applies the two sides in
    // (#1062); see `Store::merge_pins`. `take_theirs` is read from src's
    // head, not reconstructed from the session (the commit consumed it).
    let store = state.store.lock().unwrap();
    let mut entries: BTreeMap<lex_vcs::SigId, Option<lex_vcs::StageId>> =
        match store.merge_pins(dst_head.as_ref(), src_head.as_ref(), &auto_resolved, &commit_out.resolved) {
            Ok(e) => e,
            Err(e) => return write_error_response("resolve merge entries", e),
        };

    for (conflict_id, resolution) in commit_out.resolved {
        match resolution {
            // Pinned by `merge_pins` above (dst already has its head; the
            // `theirs` stage is src's).
            lex_vcs::Resolution::TakeOurs | lex_vcs::Resolution::TakeTheirs => {}
            lex_vcs::Resolution::Custom { op } => {
                // The agent's brand-new op carries the merge target
                // in its kind (e.g. ModifyBody.to_stage_id). The op
                // itself isn't separately recorded in the log here
                // — its head-map effect is folded into the merge
                // op's entries map. Callers that want the op as a
                // first-class history entry should publish it via
                // /v1/publish first and submit a TakeTheirs/TakeOurs
                // resolution against the resulting head.
                match op.kind.merge_target() {
                    Some((sig, stage)) => {
                        if sig != conflict_id {
                            return error_with_detail(
                                422,
                                "custom op targets a different sig than the conflict",
                                serde_json::json!({
                                    "conflict_id": conflict_id,
                                    "op_targets": sig,
                                }),
                            );
                        }
                        entries.insert(conflict_id, stage);
                    }
                    None => {
                        return error_with_detail(
                            422,
                            "custom op kind doesn't yield a single sig→stage delta",
                            serde_json::json!({
                                "conflict_id": conflict_id,
                                "kind": serde_json::to_value(&op.kind).unwrap_or(serde_json::Value::Null),
                            }),
                        );
                    }
                }
            }
            lex_vcs::Resolution::Defer => {
                // Unreachable: commit() rejects Defer above.
                return error_response(500, "internal: Defer slipped past commit gate");
            }
        }
    }

    let resolved_count = entries.len();
    let mut parents: Vec<lex_vcs::OpId> = Vec::new();
    if let Some(d) = dst_head.clone() { parents.push(d); }
    if let Some(s) = src_head.clone() { parents.push(s); }
    let op = lex_vcs::Operation::new(
        lex_vcs::OperationKind::Merge { resolved: resolved_count },
        parents,
    );
    let transition = lex_vcs::StageTransition::Merge { entries };

    // #1007 PR 7: if the manifests disagreed at all, build the merged
    // manifest now — auto-resolved paths plus every resolved file
    // conflict — so it can be landed as a `SetFiles` right after the
    // merge op. Recomputes the same 3-way diff `merge/start` already
    // showed the caller (over the same `lca`/`dst_head`/`src_head`, so
    // it reproduces byte-for-byte unless the store's blobs changed
    // underneath the session, which the manifest-closure check below
    // would catch anyway).
    let manifest_blob: Option<String> = if commit_out.needs_setfiles {
        match store.manifest_merge(lca.as_deref(), dst_head.as_deref(), src_head.as_deref()) {
            Ok(lex_store::ManifestMergeOutcome::NoChange) => None,
            Ok(lex_store::ManifestMergeOutcome::Needed { auto_entries, conflicts }) => {
                let file_resolutions: BTreeMap<lex_vcs::FilePath, lex_vcs::FileResolution> =
                    commit_out.resolved_files.into_iter().collect();
                match store.build_merged_manifest(
                    auto_entries,
                    &conflicts,
                    &file_resolutions,
                    "pending-merge",
                ) {
                    Ok(blob_id) => Some(blob_id),
                    Err(e) => return write_error_response("build merged manifest", e),
                }
            }
            Err(lex_store::StoreError::AmbiguousManifest { op_id }) => {
                return error_with_detail(422, "AmbiguousManifest", serde_json::json!({
                    "op_id": op_id,
                }));
            }
            Err(e) => return write_error_response("manifest merge", e),
        }
    } else {
        None
    };

    // Gated (#833): lands the merge op, type-checks the real post-merge
    // head, rolls the head back on a TypeError — and (#1007 PR 7) also
    // appends the `SetFiles` above when `manifest_blob` is `Some`,
    // rolling all the way back if that fails too (see
    // `apply_merge_op_gated_with_manifest`'s doc comment).
    match store.apply_merge_op_gated_with_manifest(&dst_branch, op, transition, manifest_blob.as_deref(), None) {
        Ok(new_head_op) => json_response(200, &serde_json::json!({
            "new_head_op": new_head_op,
            "dst_branch": dst_branch,
        })),
        Err(lex_store::StoreError::TypeError(errs)) => error_with_detail(
            422, "merged program has type errors", serde_json::to_value(&errs).unwrap_or_default()),
        // #977: the two branches pin the same dependency at different
        // versions. The merge commits the union of both parents' locks and
        // refuses to pick a side silently; nothing was written, so the dst
        // branch is unchanged. 409: the request is fine, the branches'
        // states conflict — align the version on one branch and retry.
        Err(e @ lex_store::StoreError::DependencyConflict { .. }) => {
            let detail = match &e {
                lex_store::StoreError::DependencyConflict { package, dst_version, src_version } => {
                    serde_json::json!({
                        "kind": "dependency_conflict",
                        "package": package,
                        "dst_branch": dst_branch,
                        "dst_version": dst_version,
                        "src_branch": wrapped.src_branch,
                        "src_version": src_version,
                    })
                }
                _ => serde_json::Value::Null,
            };
            error_with_detail(409, e.to_string(), detail)
        }
        Err(e @ lex_store::StoreError::AmbiguousManifest { .. }) => {
            error_with_detail(422, "AmbiguousManifest", serde_json::json!({"detail": e.to_string()}))
        }
        Err(e @ lex_store::StoreError::InvalidManifest(_))
        | Err(e @ lex_store::StoreError::MissingBlobs(_)) => {
            error_with_detail(422, e.to_string(), serde_json::Value::Null)
        }
        Err(e) => write_error_response("apply merge op", e),
    }
}

fn mint_merge_id() -> MergeSessionId {
    use std::sync::atomic::{AtomicU64, Ordering};
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("merge_{nanos:x}_{n:x}")
}

// ---- #242: append-only sync ---------------------------------------

/// `POST /v1/ops/batch` (#242). Server endpoint for `lex op push`.
///
/// Body: a JSON array of `OperationRecord`s. The handler validates
/// DAG integrity by checking that every op's `parents` either
/// already exist on the remote *or* appear earlier in the same
/// batch. This lets a client send a topologically-ordered slice
/// without first probing for what's already there.
///
/// Response shape:
///
/// ```json
/// { "received": N, "added": M, "skipped": (N-M), "added_ids": [...] }
/// ```
///
/// Failure modes:
///
/// * `400` — body isn't a JSON array of op records.
/// * `422` with `{ "error": "MissingParent", "detail": { "op_id":
///   ..., "missing_parent": ... } }` if a parent is unreachable.
///   The whole batch is rejected; nothing is persisted. The client
///   should backfill the missing op and retry.
/// * `409` if the supplied `op_id` doesn't match the canonical
///   hash of the record's payload — content addressing must hold
///   over the wire.
/// * `422` `MissingBlobs` `{ op_id, ids }` / `InvalidManifest`
///   `{ op_id, manifest, reason }` for a `SetFiles` op (#1007) whose
///   manifest or entry blobs this store lacks, or whose manifest is not
///   a valid one — so a head can never name a file set it can't serve.
///   Blobs go first (`/v1/blobs/batch`).
///
/// Idempotency: a record whose `op_id` already exists is silently
/// skipped (not added, not rejected). Pushing the same payload
/// twice is `received == N, added == 0` on the second call.
pub(crate) fn ops_batch_handler(state: &State, body: &str)
    -> Response<std::io::Cursor<Vec<u8>>>
{
    let records: Vec<lex_vcs::OperationRecord> = match serde_json::from_str(body) {
        Ok(r) => r,
        Err(e) => return error_response(400,
            format!("body must be a JSON array of OperationRecord: {e}")),
    };
    let store = state.store.lock().unwrap();
    let log = match lex_vcs::OpLog::open(store.root()) {
        Ok(l) => l,
        Err(e) => return error_response(500, format!("opening op log: {e}")),
    };

    // Validate every record before persisting any of them.
    //
    // 1. Content-addressing: the supplied `op_id` must match the
    //    canonical hash of `record.op`. Otherwise the client is
    //    sending a forged or corrupted record.
    // 2. DAG integrity: every parent must either already exist in
    //    the local log OR appear earlier in this batch.
    let mut batch_ids: std::collections::BTreeSet<lex_vcs::OpId> =
        std::collections::BTreeSet::new();
    for rec in &records {
        let expected = rec.op.op_id();
        if expected != rec.op_id {
            return error_with_detail(409, "OpIdMismatch", serde_json::json!({
                "supplied": rec.op_id,
                "expected": expected,
            }));
        }
        for parent in &rec.op.parents {
            let known = match log.get(parent) {
                Ok(Some(_)) => true,
                Ok(None) => false,
                Err(e) => return error_response(500, format!("op log read: {e}")),
            };
            if !known && !batch_ids.contains(parent) {
                return error_with_detail(422, "MissingParent", serde_json::json!({
                    "op_id": rec.op_id,
                    "missing_parent": parent,
                }));
            }
        }
        if let Err(resp) = crate::sync_http::check_set_files(&store, state.blob_limits, rec) {
            return resp;
        }
        batch_ids.insert(rec.op_id.clone());
    }

    // Persist. `OpLog::put` is idempotent so a re-push is a no-op
    // for already-present records.
    let mut added = 0usize;
    let mut added_ids: Vec<&lex_vcs::OpId> = Vec::new();
    for rec in &records {
        let already_present = matches!(log.get(&rec.op_id), Ok(Some(_)));
        match log.put(rec) {
            Ok(()) => {
                if !already_present {
                    added += 1;
                    added_ids.push(&rec.op_id);
                }
            }
            Err(e) => return error_response(500, format!("op log write: {e}")),
        }
    }

    json_response(200, &serde_json::json!({
        "received": records.len(),
        "added": added,
        "skipped": records.len() - added,
        "added_ids": added_ids,
    }))
}

/// `POST /v1/attestations/batch` (#242). Server endpoint for `lex
/// attest push`.
///
/// Body: a JSON array of `Attestation`s. Validates that each
/// attestation's `op_id` (when set) refers to an op that already
/// exists on the remote — `attestation_id` is then re-derivable
/// from the canonical form, so cross-store dedup just works.
///
/// Response: same shape as `ops_batch_handler` but `added_ids` is
/// the list of accepted `attestation_id`s.
///
/// Failure modes:
///
/// * `400` for malformed JSON.
/// * `422` with `{ "error": "UnknownOp", "detail": { ... } }` if
///   an attestation's `op_id` references an op the remote doesn't
///   know about. Whole batch rejected.
/// * `409` `AttestationIdMismatch` if the supplied id doesn't
///   match the canonical hash.
/// * `403` `ReservedKind` if any attestation's kind is in
///   [`State::reserved_kinds`] (#1066). Whole batch rejected, nothing
///   written.
/// * `403` `ReservedProducer` if any attestation claims a
///   `produced_by.tool` in [`State::reserved_producers`]. Whole batch
///   rejected, nothing written.
///
/// Check order (deliberate; a hosted embedder's dispatcher-level guard
/// answers the same way): malformed JSON `400` first — the kind is judged
/// on the parsed `Attestation`, so nothing unparsable is ever judged —
/// then `ReservedKind`, then `ReservedProducer` (a batch violating both
/// answers `ReservedKind`), then `409`/`422` validation, then persist.
///
/// Idempotency: same as the ops endpoint — content-addressed dedup.
pub(crate) fn attestations_batch_handler(state: &State, body: &str)
    -> Response<std::io::Cursor<Vec<u8>>>
{
    let attestations: Vec<lex_vcs::Attestation> = match serde_json::from_str(body) {
        Ok(a) => a,
        Err(e) => return error_response(400,
            format!("body must be a JSON array of Attestation: {e}")),
    };
    let store = state.store.lock().unwrap();
    let log = match store.attestation_log() {
        Ok(l) => l,
        Err(e) => return error_response(500, format!("opening attestation log: {e}")),
    };
    let op_log = match lex_vcs::OpLog::open(store.root()) {
        Ok(l) => l,
        Err(e) => return error_response(500, format!("opening op log: {e}")),
    };

    // Reserved kinds first (#1066), over the WHOLE batch and on the PARSED
    // kind — the very value `log.put` serializes below — so a body that
    // smuggles a second `"kind"` tag is judged as the one that would be
    // stored. Readers such as `latest_review_verdict` key on kind, not on
    // the producer name, so the producer reservation alone is not enough.
    for att in &attestations {
        if let Some((reserved, kind)) = state.reserved_kind_claimed(att) {
            return error_with_detail(403, "ReservedKind", serde_json::json!({
                "attestation_id": att.attestation_id,
                "kind": kind,
                "reserved_kind": reserved,
                "message": "this attestation kind is reserved for the server; \
                            clients may not submit attestations of it \
                            (review verdicts go through POST /v1/review/verdict)",
            }));
        }
    }

    // Reserved producers next, over the WHOLE batch, so a mixed batch
    // (one legitimate + one reserved) is refused outright and nothing is
    // written. Server-internal writers (hosted CI, review, examples) call
    // the store directly and never come through here.
    for att in &attestations {
        if let Some(reserved) = state.reserved_producer_claimed(att) {
            return error_with_detail(403, "ReservedProducer", serde_json::json!({
                "attestation_id": att.attestation_id,
                "produced_by_tool": att.produced_by.tool,
                "reserved_tool": reserved,
                "message": "this producer name is reserved for the server; \
                            clients may not submit attestations claiming it",
            }));
        }
    }

    // Validate before persisting any record.
    for att in &attestations {
        // Content-addressing: re-derive attestation_id from the
        // payload and reject mismatches.
        let expected = lex_vcs::Attestation::with_timestamp(
            att.stage_id.clone(),
            att.op_id.clone(),
            att.intent_id.clone(),
            att.kind.clone(),
            att.result.clone(),
            att.produced_by.clone(),
            att.cost.clone(),
            att.timestamp,
        ).attestation_id;
        if expected != att.attestation_id {
            return error_with_detail(409, "AttestationIdMismatch", serde_json::json!({
                "supplied": att.attestation_id,
                "expected": expected,
            }));
        }
        // The op_id field, if set, must point at an op the remote
        // knows about. Without this check, attestations would
        // dangle into a future sync that never lands the op.
        if let Some(op_id) = &att.op_id {
            match op_log.get(op_id) {
                Ok(Some(_)) => {}
                Ok(None) => return error_with_detail(422, "UnknownOp", serde_json::json!({
                    "attestation_id": att.attestation_id,
                    "op_id": op_id,
                })),
                Err(e) => return error_response(500, format!("op log read: {e}")),
            }
        }
    }

    // Persist. `AttestationLog::put` is idempotent on
    // `attestation_id` and the by-stage index is rewritten as a
    // marker file, also idempotent.
    let mut added = 0usize;
    let mut added_ids: Vec<&lex_vcs::AttestationId> = Vec::new();
    for att in &attestations {
        let already_present = matches!(log.get(&att.attestation_id), Ok(Some(_)));
        match log.put(att) {
            Ok(()) => {
                if !already_present {
                    added += 1;
                    added_ids.push(&att.attestation_id);
                }
            }
            Err(e) => return error_response(500, format!("attestation log write: {e}")),
        }
    }

    json_response(200, &serde_json::json!({
        "received": attestations.len(),
        "added": added,
        "skipped": attestations.len() - added,
        "added_ids": added_ids,
    }))
}

/// `GET /v1/attestations/since?after-op=<op_id>&limit=<n>` (#260).
/// Mirror of `ops_since_handler` for the attestation log.
///
/// Returns attestations whose `op_id` field is reachable from
/// **any** branch's head — not just one — and not in `after_op`'s
/// ancestry. The cross-branch fan-out matches the push side:
/// attestations are stage-keyed, not branch-keyed, so a single
/// "since this op" filter is the right shape.
///
/// Attestations with `op_id: None` (e.g. `Override`,
/// `ProducerBlock`) are always included — the cutoff doesn't apply.
/// `--limit` caps the response.
pub(crate) fn attestations_since_handler(state: &State, query: &str)
    -> Response<std::io::Cursor<Vec<u8>>>
{
    let mut after_op: Option<String> = None;
    let mut limit: Option<usize> = None;
    for kv in query.split('&') {
        let Some((k, v)) = kv.split_once('=') else { continue };
        match k {
            "after-op" => after_op = Some(v.to_string()),
            "limit" => {
                limit = Some(match v.parse::<usize>() {
                    Ok(n) => n,
                    Err(_) => return error_response(400,
                        format!("limit must be a positive integer, got `{v}`")),
                });
            }
            _ => {}
        }
    }

    let store = state.store.lock().unwrap();
    let log = match store.attestation_log() {
        Ok(l) => l,
        Err(e) => return error_response(500, format!("opening attestation log: {e}")),
    };

    // Build the exclude set: every op_id reachable from `after_op`,
    // inclusive. Attestations whose op_id is in this set were
    // already known to the caller.
    let exclude: std::collections::BTreeSet<String> = match &after_op {
        None => std::collections::BTreeSet::new(),
        Some(cutoff) => {
            let op_log = match lex_vcs::OpLog::open(store.root()) {
                Ok(l) => l,
                Err(e) => return error_response(500, format!("opening op log: {e}")),
            };
            match op_log.walk_back(cutoff, None) {
                Ok(records) => records.into_iter().map(|r| r.op_id).collect(),
                Err(_) => {
                    // Cutoff op doesn't exist on this remote. Treat
                    // as "no exclude" — caller will get every
                    // attestation. They may dedup client-side.
                    std::collections::BTreeSet::new()
                }
            }
        }
    };

    let all = match log.list_all() {
        Ok(v) => v,
        Err(e) => return error_response(500, format!("listing attestations: {e}")),
    };
    let mut filtered: Vec<lex_vcs::Attestation> = all
        .into_iter()
        .filter(|a| match &a.op_id {
            Some(op_id) => !exclude.contains(op_id),
            // No op_id = doesn't participate in the cutoff; always
            // ship it on the first pull, server-side idempotency
            // dedupes on the client.
            None => true,
        })
        .collect();
    // Stable order: oldest-first by `timestamp`, then by
    // `attestation_id` for ties. Lets the client land them
    // deterministically.
    filtered.sort_by(|a, b| {
        a.timestamp.cmp(&b.timestamp)
            .then_with(|| a.attestation_id.cmp(&b.attestation_id))
    });
    if let Some(n) = limit {
        filtered.truncate(n);
    }

    json_response(200, &serde_json::to_value(&filtered).unwrap_or_default())
}

// ── Package concept (#4) ────────────────────────────────────────────────────

/// Full coordinates of one declared dependency, captured from the releaser's
/// `lex.toml` so the rendered archive can reproduce a faithful `[dependencies]`
/// table — carrying BOTH a vcs (`registry` + `version`) reference and a `git`
/// mirror when the source declared both. Every field is optional so a bare
/// registry dep, a bare git dep, or a dual dep all round-trip. `#[serde(default,
/// skip_serializing_if)]` keeps the record compact and back-compatible.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
struct DepSpec {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    registry: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    version: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    git: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    branch: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    tag: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    rev: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    path: Option<String>,
}

impl DepSpec {
    /// Render this spec as the body of a `lex.toml` inline dependency table,
    /// e.g. `{ registry = "…", version = "^0.9", git = "…" }`. Keys are emitted
    /// in a stable order (vcs primary first, then the git mirror) so the
    /// rendered manifest is deterministic.
    fn to_toml_inline(&self) -> Option<String> {
        let mut parts: Vec<String> = Vec::new();
        let mut push = |k: &str, v: &Option<String>| {
            if let Some(val) = v {
                parts.push(format!("{k} = {}", toml_str(val)));
            }
        };
        push("registry", &self.registry);
        push("version", &self.version);
        push("git", &self.git);
        push("branch", &self.branch);
        push("tag", &self.tag);
        push("rev", &self.rev);
        push("path", &self.path);
        if parts.is_empty() {
            None
        } else {
            Some(format!("{{ {} }}", parts.join(", ")))
        }
    }
}

/// Quote a string as a TOML basic string (escaping `\` and `"`).
fn toml_str(s: &str) -> String {
    format!("\"{}\"", s.replace('\\', "\\\\").replace('"', "\\\""))
}

/// Per-version record stored at `{store_root}/packages/{name}/{version}.json`.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct PkgRecord {
    name: String,
    version: String,
    head_op: Option<String>,
    published_at: u64,
    /// Function names introduced or updated by this version (for retract).
    function_names: Vec<String>,
    /// External package dependencies of this release — declared by the
    /// releaser (from `lex.toml`) and/or extracted from the head's
    /// non-inlined imports. The edge set of the dependency graph (#893
    /// propagation). `#[serde(default)]` keeps pre-existing records readable.
    #[serde(default)]
    dependencies: Vec<String>,
    /// Full dependency coordinates (name → git+vcs refs) captured from the
    /// releaser's `lex.toml`, so the rendered install archive reproduces a
    /// faithful `[dependencies]` table (with both git and vcs references).
    /// `#[serde(default)]` keeps pre-existing records (which lack it) readable;
    /// when empty, the archive falls back to a bare `[package]` manifest.
    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
    dependency_specs: std::collections::BTreeMap<String, DepSpec>,
    /// Raw op JSON from each file in this publish.
    ops: Vec<serde_json::Value>,
}

/// Whether a package is reachable without authentication.
///
/// Per-package (applies across all versions), GitHub-style: an org's
/// store can hold a mix of public and private packages. Defaults to
/// `Private`, so an index written before this field existed (or any
/// freshly published package) is private until an owner opts in.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, Default)]
#[serde(rename_all = "lowercase")]
pub enum Visibility {
    #[default]
    Private,
    Public,
}

/// Index stored at `{store_root}/packages/{name}/index.json`.
///
/// Tracks which versions have been published and which is "latest", so
/// consumers can resolve `{name}@latest` without listing every file.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)]
struct PkgIndex {
    /// The most recently published version string (human label, not OpId).
    latest: Option<String>,
    /// All published versions, newest-last.
    versions: Vec<PkgVersionSummary>,
    /// Public/private flag. `#[serde(default)]` keeps pre-existing
    /// `index.json` files (which lack the field) deserializing as
    /// `Private`.
    #[serde(default)]
    visibility: Visibility,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct PkgVersionSummary {
    version: String,
    head_op: Option<String>,
    published_at: u64,
}

fn pkg_name_dir(root: &std::path::Path, name: &str) -> PathBuf {
    root.join("packages").join(name)
}

fn pkg_index_path(root: &std::path::Path, name: &str) -> PathBuf {
    pkg_name_dir(root, name).join("index.json")
}

fn pkg_version_path(root: &std::path::Path, name: &str, version: &str) -> PathBuf {
    pkg_name_dir(root, name).join(format!("{version}.json"))
}

fn pkg_archive_path(root: &std::path::Path, name: &str, version: &str) -> PathBuf {
    pkg_name_dir(root, name).join(format!("{version}.tar.gz"))
}

fn load_pkg_index(root: &std::path::Path, name: &str) -> Option<PkgIndex> {
    let bytes = std::fs::read(pkg_index_path(root, name)).ok()?;
    serde_json::from_slice(&bytes).ok()
}

fn load_pkg_record(root: &std::path::Path, name: &str, version: &str) -> Option<PkgRecord> {
    let bytes = std::fs::read(pkg_version_path(root, name, version)).ok()?;
    serde_json::from_slice(&bytes).ok()
}

fn load_latest_pkg_record(root: &std::path::Path, name: &str) -> Option<PkgRecord> {
    let index = load_pkg_index(root, name)?;
    let latest = index.latest.clone()?;
    load_pkg_record(root, name, &latest)
}

/// A package is public iff its index exists and is marked `Public`.
/// A missing index (unknown package) is treated as private, so the
/// public surface never distinguishes "private" from "does not exist".
fn pkg_is_public(root: &std::path::Path, name: &str) -> bool {
    load_pkg_index(root, name).map(|i| i.visibility) == Some(Visibility::Public)
}

/// Reject package/version path segments that could escape the
/// `packages/` directory or otherwise aren't valid names. Mirrors the
/// tenant-id guard's spirit (defense in depth — lex-hub validates the
/// tenant, this validates the package/version).
fn valid_pkg_segment(s: &str) -> bool {
    !s.is_empty()
        && s.len() <= 128
        && s != "."
        && s != ".."
        && s.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
}

#[derive(Deserialize)]
struct VisibilityReq {
    visibility: Visibility,
}

/// `PUT /v1/pkg/{name}/visibility` — set a package public or private.
///
/// Authorization is implicit: this runs against a single tenant's store,
/// and the caller only reaches *this* store because the front door
/// (lex-hub) authenticated their token and selected it. A caller can
/// therefore only change visibility of packages they own.
fn pkg_set_visibility_handler(
    state: &State,
    name: &str,
    body: &str,
) -> Response<std::io::Cursor<Vec<u8>>> {
    if !valid_pkg_segment(name) {
        return error_response(400, format!("invalid package name {name:?}"));
    }
    let req: VisibilityReq = match serde_json::from_str(body) {
        Ok(r) => r,
        Err(e) => return error_response(400, format!("bad request: {e}")),
    };
    let mut index = match load_pkg_index(&state.root, name) {
        Some(i) => i,
        None => return error_response(404, format!("package {name:?} not found")),
    };
    index.visibility = req.visibility;
    let bytes = serde_json::to_vec_pretty(&index).unwrap_or_default();
    match std::fs::write(pkg_index_path(&state.root, name), bytes) {
        Ok(()) => json_response(
            200,
            &serde_json::json!({ "name": name, "visibility": index.visibility }),
        ),
        Err(e) => error_response(500, format!("write index: {e}")),
    }
}

#[derive(serde::Deserialize)]
struct ReleaseReq {
    version: String,
    #[serde(default)]
    branch: Option<String>,
    /// External package dependencies (from the releaser's `lex.toml`).
    /// Unioned with any non-inlined external imports found in the head.
    #[serde(default)]
    dependencies: Vec<String>,
    /// Full dependency coordinates (name → git+vcs refs) from the releaser's
    /// `lex.toml`, so the rendered install archive reproduces a faithful
    /// `[dependencies]` table. Optional (back-compatible); when omitted, the
    /// archive manifest carries only `[package]`.
    #[serde(default)]
    dependency_specs: std::collections::BTreeMap<String, DepSpec>,
}

/// `POST /v1/pkg/{name}/release` — cut an immutable versioned release of
/// an **op-log-hosted** package: snapshot the current branch head as
/// `name@version` in the registry (#893). Unlike `POST /v1/pkg/publish`
/// (archive upload), this records only the op-log ref (`head_op`) — a
/// consumer resolves the version, `op pull`s that head, and renders
/// source. A published version is immutable: re-releasing an existing
/// version is a 409, so a resolved+locked dependency can never change
/// under a consumer.
fn pkg_release_handler(state: &State, name: &str, body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    if !valid_pkg_segment(name) {
        return error_response(400, format!("invalid package name {name:?}"));
    }
    let req: ReleaseReq = match serde_json::from_str(body) {
        Ok(r) => r,
        Err(e) => return error_response(400, format!("bad request: {e}")),
    };
    let version = req.version.trim().to_string();
    if version.is_empty() || !valid_pkg_segment(&version) {
        return error_response(400, "version must be a non-empty, path-safe string (e.g. 1.2.0)");
    }
    // Immutable: a published version can never be overwritten.
    if load_pkg_record(&state.root, name, &version).is_some() {
        return error_response(
            409,
            format!("{name}@{version} already released; releases are immutable — bump the version"),
        );
    }

    let store = state.store.lock().unwrap();
    let branch = req.branch.unwrap_or_else(|| store.current_branch());
    let head_op = match store.get_branch(&branch) {
        Ok(Some(b)) => b.head_op,
        Ok(None) => return error_response(404, format!("unknown branch {branch:?}")),
        Err(e) => return error_response(500, format!("get_branch: {e}")),
    };
    let Some(head_op) = head_op else {
        return error_response(400, format!("branch {branch:?} has no commits to release"));
    };

    // Version-bump gate (#893): the version increment must be at least what
    // the public-API change requires — a breaking change (a removed or
    // re-signatured public declaration) needs a *major* bump, an addition at
    // least a *minor*. This is what makes `^`/`~` resolution safe: a caret
    // update can't silently pull a breaking change mislabeled as a patch.
    //
    // Compare against the **semver predecessor** — the highest already-published
    // version strictly less than the new one (not merely `latest`, so releasing
    // 2.0.0 after 1.5.0 diffs against 1.5.0). No predecessor (first version, or
    // a back-port below everything) means nothing to gate.
    let predecessor = load_pkg_index(&state.root, name)
        .map(|i| i.versions)
        .unwrap_or_default()
        .into_iter()
        .filter_map(|v| lex_syntax::semver::parse_exact(&v.version).map(|p| (p, v)))
        .filter(|(p, _)| lex_syntax::semver::parse_exact(&version).map(|n| *p < n).unwrap_or(false))
        .max_by_key(|(p, _)| *p)
        .map(|(_, v)| v);
    if let Some(prev) = predecessor {
        if let (Some(prev_head), Some(declared)) = (
            prev.head_op.clone(),
            lex_syntax::semver::bump_between(&prev.version, &version),
        ) {
            if let (Ok(prev_api), Ok(new_api)) = (
                lex_store::api::public_api_at_op(&store, &prev_head),
                lex_store::api::public_api_at_op(&store, &head_op),
            ) {
                use lex_store::api::ApiChange;
                use lex_syntax::semver::Bump;
                let (required, why) = match lex_store::api::classify_api_change(&prev_api, &new_api) {
                    ApiChange::Breaking(d) => (Bump::Major, d),
                    ApiChange::Additive(d) => (Bump::Minor, d),
                    ApiChange::None => (Bump::Patch, String::new()),
                };
                if declared < required {
                    let need = match required {
                        Bump::Major => "major",
                        Bump::Minor => "minor",
                        Bump::Patch => "patch",
                    };
                    return error_response(
                        422,
                        format!(
                            "version bump too small: {} → {version} is a {declared:?} bump, \
                             but the API change ({why}) requires a {need} bump",
                            prev.version
                        ),
                    );
                }
            }
        }
    }

    // The package's exported function names at this head (for retract /
    // the catalog), read through the SigId the head names each by.
    let head = store.branch_head(&branch).unwrap_or_default();
    let pairs: Vec<(String, String)> = head.iter().map(|(s, st)| (s.clone(), st.clone())).collect();
    let function_names: Vec<String> = store
        .get_asts_for_sigs_bulk(&pairs)
        .into_iter()
        .filter_map(|r| r.ok())
        .filter_map(|s| match s {
            lex_ast::Stage::FnDecl(fd) => Some(fd.name),
            _ => None,
        })
        .collect();
    // Dependency-graph edges (#893 propagation): the releaser's declared deps
    // (from lex.toml) unioned with any external imports still visible in the
    // head (most are inlined at publish, so the declared list is primary).
    let mut deps: std::collections::BTreeSet<String> = req.dependencies.into_iter().collect();
    if let Ok(extracted) = lex_store::api::external_dependencies_at_op(&store, &head_op) {
        deps.extend(extracted);
    }
    // Names captured as full coordinates count as dependency-graph edges too,
    // so a release that sends only `dependency_specs` still records the edges.
    deps.extend(req.dependency_specs.keys().cloned());
    let dependencies: Vec<String> = deps.into_iter().collect();
    drop(store);

    let published_at = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let record = PkgRecord {
        name: name.to_string(),
        version: version.clone(),
        head_op: Some(head_op.clone()),
        published_at,
        function_names,
        dependencies,
        dependency_specs: req.dependency_specs,
        ops: Vec::new(),
    };
    if let Err(e) = save_pkg_record(&state.root, &record, None) {
        return error_response(500, format!("write release: {e}"));
    }
    json_response(
        201,
        &serde_json::json!({
            "name": name,
            "version": version,
            "head_op": head_op,
            "branch": branch,
        }),
    )
}

/// Names of the tenant's PUBLIC packages, sorted. Pure (filesystem in,
/// names out) so the visibility filter is unit-testable.
fn public_pkg_names(root: &std::path::Path) -> Vec<String> {
    list_pkg_names(root)
        .into_iter()
        .filter(|name| pkg_is_public(root, name))
        .collect()
}

/// `GET /v1/public/<tenant>` (org page) — list only the tenant's PUBLIC
/// packages (latest version of each). Private packages are omitted, so
/// their existence is not revealed.
fn public_pkg_list_handler(state: &State) -> Response<std::io::Cursor<Vec<u8>>> {
    let packages: Vec<serde_json::Value> = public_pkg_names(&state.root)
        .iter()
        .filter_map(|name| {
            let r = load_latest_pkg_record(&state.root, name)?;
            Some(serde_json::json!({
                "name": r.name,
                "version": r.version,
                "head_op": r.head_op,
                "published_at": r.published_at,
            }))
        })
        .collect();
    json_response(200, &serde_json::json!({ "packages": packages }))
}

/// A resolved public read target. Separated from response formatting so
/// the routing/guard logic is unit-testable without constructing HTTP
/// responses. `Err(status)` is a guard failure (405 = non-GET,
/// 404 = invalid/unknown route).
#[derive(Debug, PartialEq, Eq)]
enum PublicTarget {
    List,
    Latest(String),
    Versions(String),
    ApiDiff(String),
    Head(String),
    Version(String, String),
    Archive(String, String),
}

impl PublicTarget {
    /// The package name a target refers to, if any (`List` has none).
    fn pkg_name(&self) -> Option<&str> {
        match self {
            PublicTarget::List => None,
            PublicTarget::Latest(n)
            | PublicTarget::Versions(n)
            | PublicTarget::ApiDiff(n)
            | PublicTarget::Head(n)
            | PublicTarget::Version(n, _)
            | PublicTarget::Archive(n, _) => Some(n),
        }
    }
}

/// Pure routing decision for `/v1/public/<tenant>` reads. `path` is the
/// portion after the tenant (leading `/` ok). Enforces GET-only and
/// per-segment name validation; does NOT consult the store (visibility
/// is checked by the caller, which has store access).
fn resolve_public(method: &Method, path: &str) -> Result<PublicTarget, u16> {
    if !matches!(method, Method::Get) {
        return Err(405);
    }
    let rest = path.trim_matches('/');
    if rest.is_empty() {
        return Ok(PublicTarget::List);
    }
    let segs: Vec<&str> = rest.split('/').collect();
    if !segs.iter().all(|s| valid_pkg_segment(s)) {
        return Err(404);
    }
    match segs.as_slice() {
        [n] => Ok(PublicTarget::Latest(n.to_string())),
        [n, "versions"] => Ok(PublicTarget::Versions(n.to_string())),
        [n, "api-diff"] => Ok(PublicTarget::ApiDiff(n.to_string())),
        [n, "head"] => Ok(PublicTarget::Head(n.to_string())),
        [n, v, "archive"] => Ok(PublicTarget::Archive(n.to_string(), v.to_string())),
        [n, v] => Ok(PublicTarget::Version(n.to_string(), v.to_string())),
        _ => Err(404),
    }
}

/// Unauthenticated, read-only access to **public** packages in `state`'s
/// store. `path` is the portion of the URL after `/v1/public/<tenant>`
/// (with a leading `/`); lex-hub resolves `<tenant>` → store and calls
/// this. Visibility gating, GET-only enforcement, and segment validation
/// all live here so the whole public surface is auditable in one place.
///
/// Everything served here is package-scoped — manifests and the source
/// archive of a public package's own publish — so it cannot leak code
/// from a private package that happens to share content-addressed stages
/// in the same store.
pub fn route_public(
    state: &State,
    method: &Method,
    path: &str,
    query: &str,
) -> Response<std::io::Cursor<Vec<u8>>> {
    let target = match resolve_public(method, path) {
        Ok(t) => t,
        Err(405) => return error_response(405, "public read is GET-only"),
        Err(_) => return error_response(404, "not found"),
    };
    // The org listing already filters to public packages itself.
    if let PublicTarget::List = target {
        return public_pkg_list_handler(state);
    }
    // Single 404 for both "private" and "absent" — never reveal which.
    if let Some(name) = target.pkg_name() {
        if !pkg_is_public(&state.root, name) {
            return error_response(404, format!("package {name:?} not found"));
        }
    }
    match target {
        PublicTarget::List => unreachable!("handled above"),
        PublicTarget::Latest(n) => pkg_get_handler(state, &n),
        PublicTarget::Versions(n) => pkg_versions_handler(state, &n),
        PublicTarget::ApiDiff(n) => pkg_api_diff_handler(state, &n, query),
        PublicTarget::Head(n) => pkg_head_handler(state, &n),
        PublicTarget::Version(n, v) => pkg_get_version_handler(state, &n, &v),
        PublicTarget::Archive(n, v) => pkg_archive_handler(state, &n, &v),
    }
}

fn save_pkg_record(
    root: &std::path::Path,
    record: &PkgRecord,
    // `None` for an op-log release, whose source of truth is the op-log at
    // `head_op` (pull + render); `Some` for an archive-upload publish.
    archive: Option<&[u8]>,
) -> std::io::Result<()> {
    let dir = pkg_name_dir(root, &record.name);
    std::fs::create_dir_all(&dir)?;

    // Per-version record.
    let rec_bytes = serde_json::to_vec_pretty(record).unwrap_or_default();
    std::fs::write(pkg_version_path(root, &record.name, &record.version), rec_bytes)?;

    // Archive (tar.gz) for the download endpoint, when one was uploaded.
    if let Some(archive) = archive {
        std::fs::write(pkg_archive_path(root, &record.name, &record.version), archive)?;
    }

    // Update the index.
    let mut index = load_pkg_index(root, &record.name).unwrap_or_default();
    index.latest = Some(record.version.clone());
    if !index.versions.iter().any(|v| v.version == record.version) {
        index.versions.push(PkgVersionSummary {
            version: record.version.clone(),
            head_op: record.head_op.clone(),
            published_at: record.published_at,
        });
    }
    let idx_bytes = serde_json::to_vec_pretty(&index).unwrap_or_default();
    std::fs::write(pkg_index_path(root, &record.name), idx_bytes)
}

fn list_pkg_names(root: &std::path::Path) -> Vec<String> {
    let dir = root.join("packages");
    let Ok(entries) = std::fs::read_dir(&dir) else {
        return Vec::new();
    };
    let mut names: Vec<String> = entries
        .filter_map(|e| e.ok())
        .filter(|e| e.path().is_dir())
        .filter_map(|e| e.file_name().into_string().ok())
        .collect();
    names.sort();
    names
}

fn collect_lex_files(dir: &std::path::Path, out: &mut Vec<PathBuf>) {
    let Ok(entries) = std::fs::read_dir(dir) else { return };
    let mut entries: Vec<_> = entries.filter_map(|e| e.ok()).collect();
    entries.sort_by_key(|e| e.path());
    for entry in entries {
        let path = entry.path();
        if path.is_dir() {
            collect_lex_files(&path, out);
        } else if path.extension().and_then(|x| x.to_str()) == Some("lex") {
            out.push(path);
        }
    }
}

/// `POST /v1/pkg/publish` — publish a multi-file package from a `.tar.gz`
/// archive containing `lex.toml` and `src/**/*.lex`.
fn pkg_publish_handler(state: &State, body: &[u8]) -> Response<std::io::Cursor<Vec<u8>>> {
    let tmp = match tempfile::TempDir::new() {
        Ok(t) => t,
        Err(e) => return error_response(500, format!("create temp dir: {e}")),
    };
    {
        let gz = flate2::read::GzDecoder::new(std::io::Cursor::new(body));
        let mut ar = tar::Archive::new(gz);
        if let Err(e) = ar.unpack(tmp.path()) {
            return error_response(400, format!("unpack archive: {e}"));
        }
    }

    let toml_path = tmp.path().join("lex.toml");
    if !toml_path.exists() {
        return error_response(400, "archive must contain lex.toml at root");
    }
    let manifest = match Manifest::load(&toml_path) {
        Ok(m) => m,
        Err(e) => return error_response(400, format!("lex.toml: {e}")),
    };
    let (pkg_name, pkg_version) = match &manifest.package {
        Some(m) => (m.name.clone(), m.version.clone()),
        None => return error_response(400, "lex.toml must have a [package] section"),
    };

    // Reject a same-(name, version) re-publish BEFORE any store write:
    // this ran after the publish loop, so a 409'd duplicate still
    // appended its ops to the tenant's op log (#826).
    if load_pkg_record(&state.root, &pkg_name, &pkg_version).is_some() {
        return error_response(
            409,
            format!(
                "package {pkg_name}@{pkg_version} already published; \
                 bump the version in lex.toml to publish a new release"
            ),
        );
    }

    let src_dir = tmp.path().join("src");
    if !src_dir.exists() {
        return error_response(400, "archive must contain a src/ directory");
    }
    let mut lex_files: Vec<PathBuf> = Vec::new();
    collect_lex_files(&src_dir, &mut lex_files);
    if lex_files.is_empty() {
        return error_response(400, "no .lex files found in src/");
    }

    let store = state.store.lock().unwrap();
    let branch = store.current_branch();

    // `old_fns_by_name` mirrors the branch's current function set,
    // GROUPED by name rather than collapsed to one entry per name. The
    // branch is tenant-wide and carries history: several live functions
    // can share a bare name (#818 — three unrelated `validate`s across
    // `field.lex`/`schema.lex`/`validator.lex` in a real package, all
    // published before names were file-prefixed). `SigId` disambiguates
    // them correctly — it hashes the full signature, not just the name —
    // so the bug was ever collapsing multiple SigIds sharing a name down
    // to one `FnDecl`, silently discarding the others and corrupting
    // later diffs against them. The package side can no longer produce
    // such a collision: one pass, one prefix-mangled name per
    // declaration (#828).
    //
    // Reading it once, outside any per-file loop, is what fixed #813:
    // `branch_head` walks the whole branch op history and `get_ast` is a
    // disk fetch per live function, so re-deriving this per file was
    // O(files * (history_size + live_fn_count)) — tens of minutes on a
    // tenant with 110k+ accumulated ops. Since #828 there is one pass, so
    // the shape is structural rather than a discipline to maintain.
    let old_head = match store.branch_head(&branch) {
        Ok(h) => h,
        Err(e) => return error_response(500, format!("branch_head: {e}")),
    };
    // Resolve each AST through the SigId the branch head names, never its
    // StageId: StageIds are name-independent, so two live functions
    // differing only in name share one, `stage_index` maps it to just one
    // of their sigs, and the name that lookup missed got re-reported as an
    // Add on every publish of unchanged source (#826). It also skips that
    // index (#825) — see `get_asts_for_sigs_bulk`.
    let old_pairs: Vec<(String, String)> =
        old_head.iter().map(|(sig, stage)| (sig.clone(), stage.clone())).collect();
    let mut old_fns_by_name: BTreeMap<String, Vec<lex_ast::FnDecl>> = BTreeMap::new();
    for fd in store.get_asts_for_sigs_bulk(&old_pairs)
        .into_iter()
        .filter_map(|r| r.ok())
        .filter_map(|s| match s { lex_ast::Stage::FnDecl(fd) => Some(fd), _ => None })
    {
        old_fns_by_name.entry(fd.name.clone()).or_default().push(fd);
    }
    // Old `type`s on the branch, by (mangled) name — captured too (#895).
    // Types aren't overloaded, so a plain name map needs no take_matching.
    let mut old_types_by_name: BTreeMap<String, lex_ast::TypeDecl> = BTreeMap::new();
    for td in store.get_asts_for_sigs_bulk(&old_pairs)
        .into_iter()
        .filter_map(|r| r.ok())
        .filter_map(|s| match s { lex_ast::Stage::TypeDecl(td) => Some(td), _ => None })
    {
        old_types_by_name.insert(td.name.clone(), td);
    }
    // A name-independent fingerprint of a function's *contract*
    // (effects, param types, return type, examples — everything SigId
    // hashes except the name). Used only to disambiguate when multiple
    // candidates share a bare name: if this file's own declaration
    // structurally matches exactly one of them, that's unambiguously
    // the same evolving function; otherwise it's a new, unrelated
    // declaration that happens to reuse a name used elsewhere.
    fn structural_key(fd: &lex_ast::FnDecl) -> Option<String> {
        let mut anon = fd.clone();
        anon.name = String::new();
        lex_ast::sig_id(&lex_ast::Stage::FnDecl(anon))
    }

    // Look up (and consume) the candidate in `map[name]` that matches
    // `new_fd`'s identity. One candidate is taken as unambiguous (so a
    // signature-changing edit still reads as a modification of the same
    // function); several always require an exact structural match to
    // disambiguate, and no match means this is a distinct, unrelated
    // declaration reusing a name used elsewhere — `None` rather than a
    // guess that mis-attributes history. Safe because the candidate pool
    // is complete and never grows: it is built once, up front, from the
    // whole branch.
    fn take_matching(
        map: &mut BTreeMap<String, Vec<lex_ast::FnDecl>>,
        name: &str,
        new_fd: &lex_ast::FnDecl,
    ) -> Option<lex_ast::FnDecl> {
        let candidates = map.get_mut(name)?;
        let idx = match candidates.len() {
            0 => return None,
            1 => 0,
            _ => {
                let want = structural_key(new_fd);
                candidates.iter().position(|c| structural_key(c) == want)?
            }
        };
        let matched = candidates.remove(idx);
        if candidates.is_empty() {
            map.remove(name);
        }
        Some(matched)
    }

    // ---- one pass over the whole package -------------------------
    // `load_package` merges every file in the archive into ONE program
    // through a single shared loader pass. The flattening per-file loads
    // this replaced gave each top-level file its own copy of everything
    // it imported, so a shared dependency was canonicalized,
    // type-checked, diffed and published once per importing file: the
    // real 21-file `lex-schema` package, whose `error.lex` is imported by
    // 17 of its files, produced 2,239 `FnDecl`s for 693 distinct names
    // and paid for all 2,239 (#828). Collapsing 21 `publish_program`
    // calls into one matters even more than the 3.2x itself, because each
    // call independently reads every live function on the branch and
    // walks the op log for `old_imports`.
    //
    // The cost is that nothing is published under its bare source name
    // any more: `fn validate` in `src/field.lex` is
    // `field_<hash>.validate`. That is what makes one program safe to
    // check as a unit — the checker's global scope is keyed by name, and
    // two files may each declare their own `validate` (#818) — and it is
    // the naming change #828 asks for in exchange for the single pass.
    // The archive-publish path type-checks the loaded program with no
    // dependency resolver, so it still inlines registry/git deps to stay
    // self-contained (#930). The op-log-native `op push` path publishes
    // without inlining and resolves at the gate instead.
    let loaded = match load_package(&lex_files, tmp.path(), &pkg_name, /*inline_packages=*/ true) {
        Ok(p) => p,
        Err(e) => return error_response(400, format!("load package: {e}")),
    };
    let mut stages = canonicalize_program(&loaded.program);
    // Type errors are reported for the package, not per file: every name
    // in them carries its file's mangling prefix, so the offending file
    // is still named. Nothing is published unless the whole package
    // checks, where before each file published as it was processed and a
    // later failure left the earlier files' ops applied.
    if let Err(errs) = lex_types::check_and_rewrite_program(&mut stages) {
        return error_with_detail(
            422,
            format!("type errors in package {pkg_name}"),
            serde_json::to_value(&errs).unwrap(),
        );
    }
    let new_fns = stage_fns(&stages);
    let all_function_names: Vec<String> = new_fns.keys().cloned().collect();

    // Resolve each declaration against the branch's current state. No
    // in-request bookkeeping is needed now: one pass means each name is
    // declared once, so there is no earlier file in this request whose
    // just-published version a later one has to diff against.
    let mut old_fns: BTreeMap<String, lex_ast::FnDecl> = BTreeMap::new();
    for (name, new_fd) in &new_fns {
        if let Some(fd) = take_matching(&mut old_fns_by_name, name, new_fd) {
            old_fns.insert(name.clone(), fd);
        }
    }
    let new_types = stage_types(&stages);
    // Only diff old types this publish also declares — one on the branch
    // but absent from this file-set is left alone (as unclaimed old fns
    // are), not read as removed.
    let old_types: BTreeMap<String, lex_ast::TypeDecl> = new_types
        .keys()
        .filter_map(|n| old_types_by_name.get(n).map(|td| (n.clone(), td.clone())))
        .collect();
    let report =
        lex_vcs::compute_diff_with_types(&old_fns, &new_fns, &old_types, &new_types, false);

    // Imports stay attributed per file — `AddImport`/`RemoveImport` carry
    // an `in_file`, and history records these same root-relative keys.
    // Each file now gets only the modules it imports itself; a flattened
    // per-file load could not tell those from its children's.
    //
    // `imports_by_file` carries each import's real `as` alias (#909/#930), so
    // a non-default `import "..." as x` round-trips as `x`.
    let mut new_imports = lex_vcs::ImportMap::new();
    for (file, modules) in &loaded.imports_by_file {
        let entry = new_imports.entry(file.clone()).or_default();
        for (reference, alias) in modules {
            entry.insert(lex_vcs::ImportRef {
                reference: reference.clone(),
                alias: alias.clone(),
            });
        }
    }

    // Record each declaration's source file (#894) so an HTTP-published
    // package de-flattens in `export-git` too, not just a CLI-published one.
    let outcome = match store.publish_program_with_intent(
        &branch,
        &stages,
        &report,
        &new_imports,
        false,
        None,
        None,
        &loaded.module_prefixes,
    ) {
        Ok(outcome) => outcome,
        Err(lex_store::StoreError::TypeError(errs)) => {
            return error_with_detail(422, "type errors", serde_json::to_value(&errs).unwrap());
        }
        Err(e) => return write_error_response("publish_program", e),
    };
    let all_ops: Vec<serde_json::Value> = match serde_json::to_value(&outcome.ops) {
        Ok(serde_json::Value::Array(arr)) => arr,
        _ => Vec::new(),
    };
    let final_head_op = outcome.head_op;

    // Deliberately no "genuinely removed" cleanup pass here. Whatever
    // remains in `old_fns_by_name` was never claimed by any file in
    // THIS archive — but the branch this walks is scoped to the whole
    // TENANT, not to this one package: a tenant that has ever published
    // more than one package (confirmed in production — `lex-schema` and
    // `lex-ocpi` share a tenant) has every other package's functions
    // sitting in `old_fns_by_name` too, forever unclaimed by any file in
    // *this* package's own archive. An earlier version of this handler
    // treated all such leftovers as "removed" and would have emitted
    // RemoveFunction ops for a completely unrelated package's functions
    // on every single publish. Caught before it shipped (`diff_to_ops`
    // failed atomically on a stale SigId before applying anything, so
    // no data was actually lost) — see alpibrusl/lex-lang#818's
    // follow-up. Nothing here currently tracks which package "owns" a
    // given branch function, so there's no reliable way to tell a
    // genuine same-package removal from another package's untouched
    // function; leaving a deleted function's stage un-removed (it just
    // sits there, unreferenced) is the safe default until package-scoped
    // ownership is tracked, not silently deleting a stranger's data.
    // Since #828 the same applies to a package's own previous names: the
    // first publish after file-prefixed naming landed leaves the bare
    // names it used to publish under sitting unreferenced, for the same
    // reason — this view cannot tell them from another package's.

    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    // External deps from the head's non-inlined imports (archive publish; the
    // op-log release route also accepts declared deps from lex.toml).
    let dependencies: Vec<String> = final_head_op
        .as_ref()
        .and_then(|h| lex_store::api::external_dependencies_at_op(&store, h).ok())
        .unwrap_or_default();
    let record = PkgRecord {
        name: pkg_name.clone(),
        version: pkg_version,
        head_op: final_head_op.clone(),
        published_at: now,
        function_names: all_function_names,
        dependencies,
        // The archive-upload path stores the uploaded archive verbatim (its
        // lex.toml already carries the full dependency table), so it is served
        // directly rather than re-rendered — no captured specs needed here.
        dependency_specs: Default::default(),
        ops: all_ops.clone(),
    };
    if let Err(e) = save_pkg_record(&state.root, &record, Some(body)) {
        return error_response(500, format!("save package index: {e}"));
    }

    json_response(200, &serde_json::json!({
        "package": pkg_name,
        "ops": all_ops,
        "head_op": final_head_op,
    }))
}

/// `GET /v1/pkg` — list packages published by this tenant (latest version of each).
fn pkg_list_handler(state: &State) -> Response<std::io::Cursor<Vec<u8>>> {
    let names = list_pkg_names(&state.root);
    let packages: Vec<serde_json::Value> = names.iter()
        .filter_map(|name| {
            let idx = load_pkg_index(&state.root, name)?;
            let latest = idx.latest.as_deref()?;
            let r = load_pkg_record(&state.root, name, latest)?;
            Some(serde_json::json!({
                "name": r.name,
                "version": r.version,
                "head_op": r.head_op,
                "published_at": r.published_at,
            }))
        })
        .collect();
    json_response(200, &serde_json::json!({ "packages": packages }))
}

/// `GET /v1/pkg/{name}` — latest version details for a package.
fn pkg_get_handler(state: &State, name: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    match load_latest_pkg_record(&state.root, name) {
        Some(r) => json_response(200, &serde_json::json!({
            "name": r.name,
            "version": r.version,
            "head_op": r.head_op,
            "published_at": r.published_at,
            "function_names": r.function_names,
            "ops": r.ops,
        })),
        None => error_response(404, format!("package {name:?} not found")),
    }
}

/// `GET /v1/pkg/{name}/versions` — all published versions for a package.
fn pkg_versions_handler(state: &State, name: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    match load_pkg_index(&state.root, name) {
        Some(idx) => json_response(200, &serde_json::json!({
            "name": name,
            "latest": idx.latest,
            "versions": idx.versions,
        })),
        None => error_response(404, format!("package {name:?} not found")),
    }
}

/// `GET /v1/pkg/{name}/api-diff?from=<v>&to=<v>` — how the public API changed
/// between two releases (#893 propagation): the classification and the
/// mechanically-propagatable renames, so `lex propagate` can auto-derive its
/// `--rename` edits from a hosted release pair rather than have them restated.
fn pkg_api_diff_handler(state: &State, name: &str, query: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let mut from: Option<String> = None;
    let mut to: Option<String> = None;
    for kv in query.split('&') {
        match kv.split_once('=') {
            Some(("from", v)) => from = Some(v.to_string()),
            Some(("to", v)) => to = Some(v.to_string()),
            _ => {}
        }
    }
    let (Some(from), Some(to)) = (from, to) else {
        return error_response(400, "api-diff requires ?from=<version>&to=<version>");
    };
    let head_of = |v: &str| load_pkg_record(&state.root, name, v).and_then(|r| r.head_op);
    let (Some(from_head), Some(to_head)) = (head_of(&from), head_of(&to)) else {
        return error_response(404, format!("{name}: unknown release in {from}..{to}"));
    };

    let store = state.store.lock().unwrap();
    let (prev_api, new_api) = match (
        lex_store::api::public_api_at_op(&store, &from_head),
        lex_store::api::public_api_at_op(&store, &to_head),
    ) {
        (Ok(a), Ok(b)) => (a, b),
        _ => return error_response(500, "could not read package APIs for the given releases"),
    };
    let (change, detail) = match lex_store::api::classify_api_change(&prev_api, &new_api) {
        lex_store::api::ApiChange::Breaking(d) => ("breaking", d),
        lex_store::api::ApiChange::Additive(d) => ("additive", d),
        lex_store::api::ApiChange::None => ("none", String::new()),
    };
    let renames = lex_store::api::detect_renames(&prev_api, &new_api);
    json_response(200, &serde_json::json!({
        "name": name, "from": from, "to": to,
        "change": change, "detail": detail, "renames": renames,
    }))
}

/// `GET /v1/pkg/{name}/{version}` — specific version details.
fn pkg_get_version_handler(state: &State, name: &str, version: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    match load_pkg_record(&state.root, name, version) {
        Some(r) => json_response(200, &serde_json::json!({
            "name": r.name,
            "version": r.version,
            "head_op": r.head_op,
            "published_at": r.published_at,
            "function_names": r.function_names,
            "dependencies": r.dependencies,
            "ops": r.ops,
        })),
        None => error_response(404, format!("package {name:?}@{version:?} not found")),
    }
}

/// `GET /v1/pkg/{name}/{version}/archive` — download the source tar.gz.
fn pkg_archive_handler(state: &State, name: &str, version: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let gzip = |bytes: Vec<u8>| {
        Response::from_data(bytes).with_status_code(200).with_header(
            tiny_http::Header::from_bytes(&b"Content-Type"[..], &b"application/gzip"[..]).unwrap(),
        )
    };

    // 1. A stored archive (the `lex pkg publish --registry` upload path).
    if let Ok(bytes) = std::fs::read(pkg_archive_path(&state.root, name, version)) {
        return gzip(bytes);
    }

    // 2. An op-log-native release (op push + `POST …/release`, #911) has no
    //    stored archive — render one from the pinned op-log head so the
    //    package installs like any other (#920).
    if let Some(record) = load_pkg_record(&state.root, name, version) {
        if let Some(head_op) = record.head_op.clone() {
            match render_op_log_archive(state, name, version, &head_op, &record.dependency_specs) {
                Ok(bytes) => return gzip(bytes),
                Err(e) => {
                    return error_response(500, format!("rendering archive for {name:?}@{version:?}: {e}"));
                }
            }
        }
    }

    error_response(404, format!("archive for {name:?}@{version:?} not found"))
}

/// The archive path for a single-module head that records no source file of
/// its own — every op predating `in_file`. Historical, and kept so already
/// released packages in that state do not change name underneath their
/// dependents; the cure for them is a re-push, not a rename here (#988).
const DEFAULT_ARCHIVE_MODULE: &str = "src/lib.lex";

/// Build a gzip-tar package archive (`lex.toml` + the package's `src/*.lex`)
/// by rendering the op-log head `head_op` to source — the composition of a
/// registry release (#911) with the op-log, so an `op push`-hosted package is
/// installable without a separately-uploaded archive (#920).
fn render_op_log_archive(
    state: &State,
    name: &str,
    version: &str,
    head_op: &str,
    dependency_specs: &std::collections::BTreeMap<String, DepSpec>,
) -> Result<Vec<u8>, String> {
    // De-flatten the head into its source tree — one file for a single-module
    // package, or the full `src/*.lex` layout for a multi-module one (#894).
    // The same renderer `lex export-git` uses, so the installed source matches
    // the git mirror. Either way the *names* come from the head itself; this
    // used to invent `src/lib.lex` for the single-module arm (#988).
    let (files, lock): (Vec<(String, String)>, Option<String>) = {
        let store = state.store.lock().unwrap();
        // #943: ship the lock this head was published against, so a consumer
        // that installs this package resolves ITS dependencies against the
        // pins it was built with. Without it a cached package directory has
        // no `lex.lock`, and any caret registry dependency of a dependency
        // was unresolvable (`UnlockedRegistryDep`).
        let lock = store.committed_lock_inherited(head_op).ok().flatten();
        let head = lex_store::render::package_head_at_op(&store, head_op)
            .map_err(|e| format!("reading head {head_op}: {e}"))?;
        let files = match lex_store::render::render_source(&store, &head)
            .map_err(|e| format!("rendering source at {head_op}: {e}"))?
        {
            lex_store::render::RenderedSource::Single { path, src } => {
                // A head that records no file keeps the registry's historical
                // name; one that does keeps its own (#988).
                vec![(path.unwrap_or_else(|| DEFAULT_ARCHIVE_MODULE.to_string()), src)]
            }
            lex_store::render::RenderedSource::Multi(tree) => tree.into_iter().collect(),
        };
        (files, lock)
    };

    // Reconstruct the manifest from the release record. When the release
    // captured full dependency coordinates, emit a faithful `[dependencies]`
    // table (carrying both git and vcs references where the source declared
    // both); otherwise a bare `[package]` (pre-dependency-capture releases).
    let mut manifest = format!("[package]\nname = \"{name}\"\nversion = \"{version}\"\n");
    let dep_lines: Vec<String> = dependency_specs
        .iter()
        .filter_map(|(dep_name, spec)| spec.to_toml_inline().map(|inline| format!("{dep_name} = {inline}")))
        .collect();
    if !dep_lines.is_empty() {
        manifest.push_str("\n[dependencies]\n");
        for line in dep_lines {
            manifest.push_str(&line);
            manifest.push('\n');
        }
    }
    let mut enc = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
    {
        let mut ar = tar::Builder::new(&mut enc);
        let mut append = |p: &str, data: &[u8]| -> std::io::Result<()> {
            let mut h = tar::Header::new_gnu();
            h.set_size(data.len() as u64);
            h.set_mode(0o644);
            h.set_cksum();
            ar.append_data(&mut h, p, data)
        };
        append("lex.toml", manifest.as_bytes()).map_err(|e| e.to_string())?;
        if let Some(lock) = &lock {
            append("lex.lock", lock.as_bytes()).map_err(|e| e.to_string())?;
        }
        for (path, src) in &files {
            append(path, src.as_bytes()).map_err(|e| e.to_string())?;
        }
        ar.finish().map_err(|e| e.to_string())?;
    }
    enc.finish().map_err(|e| e.to_string())
}

/// `GET /v1/pkg/{name}/head` — head op for a package's latest version.
fn pkg_head_handler(state: &State, name: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    match load_latest_pkg_record(&state.root, name) {
        Some(r) => json_response(200, &serde_json::json!({
            "name": r.name,
            "version": r.version,
            "head_op": r.head_op,
        })),
        None => error_response(404, format!("package {name:?} not found")),
    }
}

/// `DELETE /v1/pkg/{name}` — retract the latest version of a package.
fn pkg_delete_handler(state: &State, name: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let record = match load_latest_pkg_record(&state.root, name) {
        Some(r) => r,
        None => return error_response(404, format!("package {name:?} not found")),
    };

    let store = state.store.lock().unwrap();
    let branch = store.current_branch();

    let head = match store.branch_head(&branch) {
        Ok(h) => h,
        Err(e) => return error_response(500, format!("branch_head: {e}")),
    };

    // Build old_fns from this package's function names that are still on
    // the branch, reading each AST through the SigId the head names. A
    // StageId-keyed read is ambiguous when two live functions differ only
    // in name (#826), and here that ambiguity decides what gets REMOVED:
    // it could both miss one of this package's functions and match a name
    // belonging to another package sharing the stage.
    let head_pairs: Vec<(String, String)> = head
        .iter()
        .map(|(sig, stage)| (sig.clone(), stage.clone()))
        .collect();
    let old_fns: BTreeMap<String, lex_ast::FnDecl> = store
        .get_asts_for_sigs_bulk(&head_pairs)
        .into_iter()
        .filter_map(|r| r.ok())
        .filter_map(|s| match s {
            lex_ast::Stage::FnDecl(fd)
                if record.function_names.contains(&fd.name) => Some((fd.name.clone(), fd)),
            _ => None,
        })
        .collect();

    let new_fns: BTreeMap<String, lex_ast::FnDecl> = BTreeMap::new();
    let report = lex_vcs::compute_diff(&old_fns, &new_fns, false);
    let empty_imports = lex_vcs::ImportMap::new();

    match store.publish_program(&branch, &[], &report, &empty_imports, false) {
        Ok(outcome) => {
            // Remove the version record and archive, then update the index.
            let ver = record.version.clone();
            let _ = std::fs::remove_file(pkg_version_path(&state.root, name, &ver));
            let _ = std::fs::remove_file(pkg_archive_path(&state.root, name, &ver));
            // Update index: remove this version, set latest to previous if any.
            if let Some(mut idx) = load_pkg_index(&state.root, name) {
                idx.versions.retain(|v| v.version != ver);
                idx.latest = idx.versions.last().map(|v| v.version.clone());
                if idx.versions.is_empty() {
                    let _ = std::fs::remove_dir_all(pkg_name_dir(&state.root, name));
                } else {
                    let bytes = serde_json::to_vec_pretty(&idx).unwrap_or_default();
                    let _ = std::fs::write(pkg_index_path(&state.root, name), bytes);
                }
            }
            json_response(200, &serde_json::json!({
                "deleted": name,
                "version": ver,
                "ops": outcome.ops,
                "head_op": outcome.head_op,
            }))
        }
        Err(lex_store::StoreError::TypeError(errs)) => {
            error_with_detail(422, "type errors", serde_json::to_value(&errs).unwrap())
        }
        Err(e) => write_error_response("retract package", e),
    }
}

#[cfg(test)]
mod dep_spec_tests {
    use super::DepSpec;

    #[test]
    fn a_dual_spec_renders_both_git_and_vcs_refs_vcs_first() {
        let spec = DepSpec {
            registry: Some("vcs.lexlang.org/lex-official/lex-schema".into()),
            version: Some("^0.9".into()),
            git: Some("https://github.com/alpibrusl/lex-schema".into()),
            ..Default::default()
        };
        assert_eq!(
            spec.to_toml_inline().as_deref(),
            Some("{ registry = \"vcs.lexlang.org/lex-official/lex-schema\", version = \"^0.9\", git = \"https://github.com/alpibrusl/lex-schema\" }"),
        );
    }

    #[test]
    fn bare_git_and_bare_registry_specs_render_their_own_keys() {
        let git = DepSpec { git: Some("https://x/g".into()), tag: Some("v1".into()), ..Default::default() };
        assert_eq!(git.to_toml_inline().as_deref(), Some("{ git = \"https://x/g\", tag = \"v1\" }"));
        let reg = DepSpec { registry: Some("vcs/r".into()), version: Some("1.0.0".into()), ..Default::default() };
        assert_eq!(reg.to_toml_inline().as_deref(), Some("{ registry = \"vcs/r\", version = \"1.0.0\" }"));
    }

    #[test]
    fn an_empty_spec_renders_nothing() {
        assert_eq!(DepSpec::default().to_toml_inline(), None);
    }
}

#[cfg(test)]
mod policy_ceiling_tests {
    use super::*;
    use lex_runtime::Policy;
    use std::path::PathBuf;

    /// A maximally-permissive policy of the kind a malicious caller
    /// would put in a `/v1/run` body: every dangerous effect plus
    /// fs over `/`.
    fn permissive_request() -> Policy {
        Policy {
            allow_effects: ["io", "fs_read", "fs_write", "net", "proc"]
                .iter()
                .map(|s| s.to_string())
                .collect(),
            allow_fs_read: vec![PathBuf::from("/")],
            allow_fs_write: vec![PathBuf::from("/")],
            allow_net_host: Vec::new(),
            allow_proc: Vec::new(),
            allow_approval: Vec::new(),
            budget: None,
        }
    }

    #[test]
    fn ceiling_drops_effects_the_caller_was_not_granted() {
        let ceiling = Policy {
            allow_effects: ["io", "time"].iter().map(|s| s.to_string()).collect(),
            ..Policy::default()
        };
        let got = clamp_policy(permissive_request(), &ceiling);
        assert!(got.allow_effects.contains("io"));
        assert!(!got.allow_effects.contains("proc"), "proc must not survive a ceiling without it");
        assert!(!got.allow_effects.contains("fs_write"));
        assert!(!got.allow_effects.contains("net"));
        // `time` is in the ceiling but not the request → intersection drops it.
        assert!(!got.allow_effects.contains("time"));
    }

    #[test]
    fn ceiling_scopes_override_caller_scopes() {
        let ceiling = Policy {
            allow_effects: ["fs_read"].iter().map(|s| s.to_string()).collect(),
            allow_fs_read: vec![PathBuf::from("/srv/tenant")],
            ..Policy::default()
        };
        let got = clamp_policy(permissive_request(), &ceiling);
        // Caller asked for "/" but only the ceiling's scope survives —
        // an empty/wider caller list can never widen the ceiling.
        assert_eq!(got.allow_fs_read, vec![PathBuf::from("/srv/tenant")]);
        assert!(got.allow_fs_write.is_empty());
        assert!(got.allow_proc.is_empty());
        assert!(got.allow_net_host.is_empty());
    }

    #[test]
    fn ceiling_caps_budget_and_prefers_the_smaller() {
        // Caller wants unlimited; ceiling caps it.
        let mut req = permissive_request();
        req.budget = None;
        let ceiling = Policy { budget: Some(1_000), ..Policy::default() };
        assert_eq!(clamp_policy(req, &ceiling).budget, Some(1_000));

        // Caller asks for less than the ceiling → keep the caller's.
        let mut req2 = permissive_request();
        req2.budget = Some(50);
        let ceiling2 = Policy { budget: Some(1_000), ..Policy::default() };
        assert_eq!(clamp_policy(req2, &ceiling2).budget, Some(50));
    }

    #[test]
    fn empty_ceiling_is_pure_only() {
        let got = clamp_policy(permissive_request(), &Policy::default());
        assert!(got.allow_effects.is_empty(), "an empty ceiling grants nothing");
        assert!(got.allow_proc.is_empty());
        assert!(got.allow_fs_write.is_empty());
    }
}

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

    /// Write a minimal package (index + per-version record + an archive
    /// blob) straight into a temp store, bypassing the publish pipeline.
    fn seed_pkg(root: &std::path::Path, name: &str, version: &str) {
        let record = PkgRecord {
            name: name.to_string(),
            version: version.to_string(),
            head_op: Some(format!("op-{name}")),
            published_at: 1,
            function_names: vec![format!("{name}.f")],
            dependencies: vec![],
            dependency_specs: Default::default(),
            ops: vec![],
        };
        save_pkg_record(root, &record, Some(format!("ARCHIVE:{name}@{version}").as_bytes()))
            .expect("seed package");
    }

    #[test]
    fn new_package_defaults_to_private() {
        let tmp = tempfile::TempDir::new().unwrap();
        seed_pkg(tmp.path(), "lex-schema", "0.9.2");
        assert!(!pkg_is_public(tmp.path(), "lex-schema"));
        // Unknown packages are also "not public" — never distinguished.
        assert!(!pkg_is_public(tmp.path(), "does-not-exist"));
    }

    #[test]
    fn set_visibility_round_trips_and_index_persists() {
        let tmp = tempfile::TempDir::new().unwrap();
        let state = State::open(tmp.path().to_path_buf()).unwrap();
        seed_pkg(tmp.path(), "lex-schema", "0.9.2");

        let _ = pkg_set_visibility_handler(&state, "lex-schema", r#"{"visibility":"public"}"#);
        assert!(pkg_is_public(tmp.path(), "lex-schema"));
        // The version list survives the index rewrite (we don't clobber it).
        let idx = load_pkg_index(tmp.path(), "lex-schema").unwrap();
        assert_eq!(idx.latest.as_deref(), Some("0.9.2"));
        assert_eq!(idx.versions.len(), 1);

        let _ = pkg_set_visibility_handler(&state, "lex-schema", r#"{"visibility":"private"}"#);
        assert!(!pkg_is_public(tmp.path(), "lex-schema"));
    }

    #[test]
    fn set_visibility_on_unknown_package_is_a_noop() {
        let tmp = tempfile::TempDir::new().unwrap();
        let state = State::open(tmp.path().to_path_buf()).unwrap();
        // No package seeded → handler returns 404 and writes nothing.
        let _ = pkg_set_visibility_handler(&state, "ghost", r#"{"visibility":"public"}"#);
        assert!(load_pkg_index(tmp.path(), "ghost").is_none());
    }

    #[test]
    fn public_listing_omits_private_packages() {
        let tmp = tempfile::TempDir::new().unwrap();
        let state = State::open(tmp.path().to_path_buf()).unwrap();
        seed_pkg(tmp.path(), "pub-pkg", "1.0.0");
        seed_pkg(tmp.path(), "priv-pkg", "1.0.0");
        let _ = pkg_set_visibility_handler(&state, "pub-pkg", r#"{"visibility":"public"}"#);

        let names = public_pkg_names(tmp.path());
        assert_eq!(names, vec!["pub-pkg".to_string()]);
    }

    #[test]
    fn resolve_public_maps_routes() {
        let get = Method::Get;
        assert_eq!(resolve_public(&get, "").unwrap(), PublicTarget::List);
        assert_eq!(resolve_public(&get, "/").unwrap(), PublicTarget::List);
        assert_eq!(
            resolve_public(&get, "/lex-schema").unwrap(),
            PublicTarget::Latest("lex-schema".into())
        );
        assert_eq!(
            resolve_public(&get, "/lex-schema/versions").unwrap(),
            PublicTarget::Versions("lex-schema".into())
        );
        assert_eq!(
            resolve_public(&get, "/lex-schema/head").unwrap(),
            PublicTarget::Head("lex-schema".into())
        );
        assert_eq!(
            resolve_public(&get, "/lex-schema/0.9.2").unwrap(),
            PublicTarget::Version("lex-schema".into(), "0.9.2".into())
        );
        assert_eq!(
            resolve_public(&get, "/lex-schema/0.9.2/archive").unwrap(),
            PublicTarget::Archive("lex-schema".into(), "0.9.2".into())
        );
    }

    #[test]
    fn resolve_public_rejects_bad_method_and_traversal() {
        // Non-GET → 405.
        assert_eq!(resolve_public(&Method::Put, "/lex-schema"), Err(405));
        assert_eq!(resolve_public(&Method::Post, "").err(), Some(405));
        // Path traversal / invalid segments → 404, never a filesystem touch.
        assert_eq!(resolve_public(&Method::Get, "/.."), Err(404));
        assert_eq!(resolve_public(&Method::Get, "/lex-schema/../etc"), Err(404));
        assert_eq!(resolve_public(&Method::Get, "/a/b/c/d"), Err(404));
        // Slashes elsewhere can't smuggle a deep path: each segment is checked.
        assert!(resolve_public(&Method::Get, "/lex schema").is_err());
    }
}