Skip to main content

lex_api/
handlers.rs

1//! Request routing for the agent API.
2//!
3//! Each handler is a synchronous function that returns
4//! `Result<serde_json::Value, ApiError>`. The dispatcher wraps the result
5//! in an HTTP response — successes as 200 with the JSON body, structured
6//! errors as 4xx/5xx with a JSON envelope.
7
8use indexmap::IndexMap;
9use lex_ast::canonicalize_program;
10use lex_bytecode::{compile_program, vm::Vm, Value};
11use lex_runtime::{check_program as check_policy, DefaultHandler, Policy};
12use lex_store::Store;
13use crate::publish_examples::record_examples_for_publish;
14use lex_syntax::{load_package, load_program_from_str, Manifest};
15use lex_vcs::{MergeSession, MergeSessionId};
16use serde::{Deserialize, Serialize};
17use std::collections::{BTreeMap, BTreeSet, HashMap};
18use std::path::PathBuf;
19use std::sync::{Arc, Mutex};
20use std::time::{SystemTime, UNIX_EPOCH};
21use tiny_http::{Header, Method, Request, Response};
22
23/// The function declarations in a canonicalized program, by name.
24fn stage_fns(stages: &[lex_ast::Stage]) -> BTreeMap<String, lex_ast::FnDecl> {
25    stages.iter().filter_map(|s| match s {
26        lex_ast::Stage::FnDecl(fd) => Some((fd.name.clone(), fd.clone())),
27        _ => None,
28    }).collect()
29}
30
31/// The type declarations in a canonicalized program, by name — so the
32/// publish diff captures `type`s alongside functions (#895).
33fn stage_types(stages: &[lex_ast::Stage]) -> BTreeMap<String, lex_ast::TypeDecl> {
34    stages.iter().filter_map(|s| match s {
35        lex_ast::Stage::TypeDecl(td) => Some((td.name.clone(), td.clone())),
36        _ => None,
37    }).collect()
38}
39
40pub struct State {
41    pub store: Mutex<Store>,
42    /// Filesystem root of the store. Held alongside the `Store`
43    /// itself so handlers that need to read store-level files
44    /// (e.g. `users.json` for actor auth) don't have to round-
45    /// trip through the lock.
46    pub root: PathBuf,
47    /// In-memory merge sessions, keyed by MergeSessionId. Sessions
48    /// are ephemeral by design (#134 foundation): they live for the
49    /// lifetime of the server process and are GC'd on commit. A
50    /// future slice can persist them to disk so a session survives
51    /// process restarts. For now an agent that gets unlucky with a
52    /// restart re-runs `merge/start` and gets a fresh session.
53    pub sessions: Mutex<HashMap<MergeSessionId, ApiMergeSession>>,
54    /// Optional server-imposed ceiling on the effect policy honored
55    /// by `/v1/run` and `/v1/replay`. `None` (the default, used by
56    /// single-tenant `lex serve`) runs the caller's request policy
57    /// as-is — the operator *is* the caller there, so that's
58    /// intended. When `Some`, the request policy is clamped via
59    /// [`clamp_policy`] so it can only *narrow* the ceiling, never
60    /// widen it.
61    ///
62    /// Any embedder that exposes this API to untrusted callers — a
63    /// hosted, multi-tenant gateway like lex-hub — MUST set this.
64    /// Without it the request body can grant itself `[proc]`
65    /// (arbitrary subprocess spawn), `[fs_*]` over `/`, and
66    /// unrestricted `[net]`: arbitrary code execution as the server
67    /// process. See lex-hub#6.
68    ///
69    /// NOTE: an empty scope list means "any path/host" in the
70    /// runtime, so a ceiling that puts `fs_read`/`fs_write`/`net` in
71    /// `allow_effects` MUST also populate the matching scope list
72    /// (`allow_fs_read`, …) or it re-opens the wildcard. Granting
73    /// none of those kinds is the safe default.
74    pub policy_ceiling: Option<Policy>,
75}
76
77/// Server-side wrapper around [`MergeSession`] carrying the
78/// branch names that started the merge. The lex-vcs session
79/// itself only tracks `OpId` heads; commit needs the dst branch
80/// name to advance the right head, and the src branch name is
81/// kept for round-trip auditability ("which branch did we merge
82/// from?").
83pub struct ApiMergeSession {
84    pub inner: MergeSession,
85    pub src_branch: String,
86    pub dst_branch: String,
87}
88
89impl State {
90    pub fn open(root: PathBuf) -> anyhow::Result<Self> {
91        Self::open_with_ceiling(root, None)
92    }
93
94    /// Like [`State::open`] but installs a [`policy_ceiling`](State::policy_ceiling)
95    /// that `/v1/run` and `/v1/replay` clamp the caller's request
96    /// policy against. Embedders exposing this API to untrusted
97    /// callers must use this constructor (or set the field directly).
98    pub fn open_with_ceiling(
99        root: PathBuf,
100        policy_ceiling: Option<Policy>,
101    ) -> anyhow::Result<Self> {
102        Ok(Self {
103            store: Mutex::new(Store::open(&root)?),
104            root,
105            sessions: Mutex::new(HashMap::new()),
106            policy_ceiling,
107        })
108    }
109
110    /// Construct a per-tenant `State` by prefixing `store_root` with the
111    /// tenant id. Single-tenant `lex serve` is unaffected — it calls
112    /// `State::open` directly.
113    ///
114    /// `tenant_id` is restricted to `[A-Za-z0-9_-]{1,64}`: anything else
115    /// (path separators, `..`, NUL, absolute paths, dotfiles, empty
116    /// string) is rejected before touching the filesystem. Without this
117    /// `PathBuf::join("/etc")` would silently replace `store_root`, and
118    /// `PathBuf::join("../foo")` would escape the tenant root.
119    pub fn new_with_tenant(tenant_id: &str, store_root: PathBuf) -> anyhow::Result<Self> {
120        validate_tenant_id(tenant_id)?;
121        Self::open(store_root.join(tenant_id))
122    }
123
124    /// Multi-tenant constructor that also installs a policy ceiling
125    /// for `/v1/run` / `/v1/replay`. The path-traversal guard from
126    /// [`new_with_tenant`](State::new_with_tenant) and the effect
127    /// ceiling are the two halves a hosted gateway needs.
128    pub fn new_with_tenant_and_ceiling(
129        tenant_id: &str,
130        store_root: PathBuf,
131        policy_ceiling: Option<Policy>,
132    ) -> anyhow::Result<Self> {
133        validate_tenant_id(tenant_id)?;
134        Self::open_with_ceiling(store_root.join(tenant_id), policy_ceiling)
135    }
136}
137
138/// Clamp a caller-supplied [`Policy`] to a server-imposed `ceiling`
139/// so it can only *narrow* the granted capabilities, never widen
140/// them. Used by [`run_handler`] when [`State::policy_ceiling`] is
141/// set — i.e. when an embedder exposes `/v1/run` to untrusted
142/// callers and must not let the request body grant itself `[proc]`,
143/// arbitrary `[fs_*]` paths, or unrestricted `[net]`.
144///
145/// - **Effects**: set-intersection of request and ceiling. The
146///   caller may drop effects but never add one the ceiling withheld.
147/// - **Scopes** (fs paths, proc binaries, net hosts): taken from the
148///   ceiling outright. The caller cannot widen them, and — because an
149///   empty scope list means "any" in the runtime — we must not let a
150///   caller's empty list collapse the ceiling's restriction back to a
151///   wildcard.
152/// - **Budget**: the more restrictive (smaller) of the two.
153fn clamp_policy(requested: Policy, ceiling: &Policy) -> Policy {
154    let allow_effects: BTreeSet<String> = requested
155        .allow_effects
156        .intersection(&ceiling.allow_effects)
157        .cloned()
158        .collect();
159    let budget = match (requested.budget, ceiling.budget) {
160        (Some(r), Some(c)) => Some(r.min(c)),
161        (None, Some(c)) => Some(c),
162        (Some(r), None) => Some(r),
163        (None, None) => None,
164    };
165    Policy {
166        allow_effects,
167        allow_fs_read: ceiling.allow_fs_read.clone(),
168        allow_fs_write: ceiling.allow_fs_write.clone(),
169        allow_net_host: ceiling.allow_net_host.clone(),
170        allow_proc: ceiling.allow_proc.clone(),
171        allow_approval: ceiling.allow_approval.clone(),
172        budget,
173    }
174}
175
176fn validate_tenant_id(tenant_id: &str) -> anyhow::Result<()> {
177    if tenant_id.is_empty() {
178        anyhow::bail!("tenant_id must not be empty");
179    }
180    if tenant_id.len() > 64 {
181        anyhow::bail!("tenant_id must be at most 64 bytes");
182    }
183    if !tenant_id
184        .bytes()
185        .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-')
186    {
187        anyhow::bail!(
188            "tenant_id {tenant_id:?} contains characters outside [A-Za-z0-9_-]"
189        );
190    }
191    Ok(())
192}
193
194#[derive(Debug, Serialize, Deserialize)]
195struct ErrorEnvelope {
196    error: String,
197    #[serde(skip_serializing_if = "Option::is_none")]
198    detail: Option<serde_json::Value>,
199}
200
201pub(crate) fn json_response(status: u16, body: &serde_json::Value) -> Response<std::io::Cursor<Vec<u8>>> {
202    let bytes = serde_json::to_vec(body).unwrap_or_else(|_| b"{}".to_vec());
203    Response::from_data(bytes)
204        .with_status_code(status)
205        .with_header(Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..]).unwrap())
206}
207
208pub(crate) fn error_response(status: u16, msg: impl Into<String>) -> Response<std::io::Cursor<Vec<u8>>> {
209    json_response(status, &serde_json::to_value(ErrorEnvelope {
210        error: msg.into(), detail: None,
211    }).unwrap())
212}
213
214pub(crate) fn error_with_detail(status: u16, msg: impl Into<String>, detail: serde_json::Value)
215    -> Response<std::io::Cursor<Vec<u8>>>
216{
217    json_response(status, &serde_json::to_value(ErrorEnvelope {
218        error: msg.into(), detail: Some(detail),
219    }).unwrap())
220}
221
222/// The actionable next step for a refused unsatisfiable pair (#992). Shared
223/// with the `lex op push` client, which prints it verbatim.
224pub const UNSATISFIABLE_PAIR_HINT: &str =
225    "republish from source to retire the stranded entry (#995)";
226
227/// `StoreError::UnsatisfiablePair` → **422**, never 500 (#992). The request
228/// asked to move a head onto a `(sig, stage)` pair no store can hold — a
229/// problem with the client's data, not the server. The body names the pair,
230/// the sig that actually owns the stage, and what to do about it:
231///
232/// ```json
233/// { "error": "UnsatisfiablePair",
234///   "detail": { "sig_id": "...", "stage_id": "...", "filed_under": "...",
235///               "hint": "republish from source to retire the stranded entry (#995)" } }
236/// ```
237///
238/// `None` for any other error, so callers fall through to their own mapping.
239pub(crate) fn unsatisfiable_pair_response(err: &lex_store::StoreError)
240    -> Option<Response<std::io::Cursor<Vec<u8>>>>
241{
242    let lex_store::StoreError::UnsatisfiablePair { sig_id, stage_id, filed_under } = err else {
243        return None;
244    };
245    Some(error_with_detail(422, "UnsatisfiablePair", serde_json::json!({
246        "sig_id": sig_id,
247        "stage_id": stage_id,
248        "filed_under": filed_under,
249        "message": err.to_string(),
250        "hint": UNSATISFIABLE_PAIR_HINT,
251    })))
252}
253
254/// Map a `StoreError` from a write path (`apply_operation` /
255/// `apply_operation_checked`) to an HTTP response. The only special
256/// case today is `Contention` (#262 multi-writer CAS retries
257/// exhausted), which maps to 503 with a `Retry-After` header so
258/// clients back off rather than hammering the same branch tip.
259fn write_error_response(prefix: &str, err: lex_store::StoreError)
260    -> Response<std::io::Cursor<Vec<u8>>>
261{
262    if let Some(resp) = unsatisfiable_pair_response(&err) {
263        return resp;
264    }
265    if let lex_store::StoreError::Contention { branch, attempts } = &err {
266        let body = serde_json::to_vec(&ErrorEnvelope {
267            error: format!("{prefix}: branch '{branch}' is contended (attempts={attempts})"),
268            detail: Some(serde_json::json!({
269                "kind": "contention",
270                "branch": branch,
271                "attempts": attempts,
272            })),
273        }).unwrap_or_else(|_| b"{}".to_vec());
274        return Response::from_data(body)
275            .with_status_code(503)
276            .with_header(Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..]).unwrap())
277            .with_header(Header::from_bytes(&b"Retry-After"[..], &b"1"[..]).unwrap());
278    }
279    // #292 slice 3: budget overflow → 503 with `Retry-After: 0`.
280    // Unlike Contention (where a retry might land after another
281    // writer finishes), there's no point retrying a budget-
282    // exceeded op — the caller needs to raise the cap, switch
283    // sessions, or refactor the work. The `Retry-After: 0`
284    // signals "don't bother retrying as-is" while still using
285    // the canonical "service refused" status code.
286    if let lex_store::StoreError::BudgetExceeded { session_id, cap, spent_after } = &err {
287        let body = serde_json::to_vec(&ErrorEnvelope {
288            error: format!(
289                "{prefix}: session `{session_id}` budget exceeded \
290                 (spent_after={spent_after}, cap={cap})"
291            ),
292            detail: Some(serde_json::json!({
293                "kind": "budget_exceeded",
294                "session_id": session_id,
295                "cap": cap,
296                "spent_after": spent_after,
297            })),
298        }).unwrap_or_else(|_| b"{}".to_vec());
299        return Response::from_data(body)
300            .with_status_code(503)
301            .with_header(Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..]).unwrap())
302            .with_header(Header::from_bytes(&b"Retry-After"[..], &b"0"[..]).unwrap());
303    }
304    error_response(500, format!("{prefix}: {err}"))
305}
306
307pub fn handle(state: Arc<State>, mut req: Request) -> std::io::Result<()> {
308    let method = req.method().clone();
309    let url = req.url().to_string();
310    let path = url.split('?').next().unwrap_or("").to_string();
311    let query = url.split_once('?').map(|(_, q)| q.to_string()).unwrap_or_default();
312
313    // `X-Lex-User` is the v3d session identifier — set by humans
314    // operating the web UI through whatever proxy fronts auth, or
315    // by AI agents calling the JSON API. We pluck it once here so
316    // every handler can take it as a borrowed string.
317    let x_lex_user = req.headers().iter()
318        .find(|h| h.field.equiv("x-lex-user"))
319        .map(|h| h.value.as_str().to_string());
320
321    // POST /v1/pkg/publish sends a raw tar.gz body — read bytes before routing.
322    if matches!(method, Method::Post) && path == "/v1/pkg/publish" {
323        let mut body_bytes: Vec<u8> = Vec::new();
324        let _ = req.as_reader().read_to_end(&mut body_bytes);
325        let resp = pkg_publish_handler(&state, &body_bytes);
326        return req.respond(resp);
327    }
328
329    let mut body = String::new();
330    let _ = req.as_reader().read_to_string(&mut body);
331
332    let resp = route(&state, &method, &path, &query, &body, x_lex_user.as_deref());
333    req.respond(resp)
334}
335
336/// Auth-gated entry point. Calls `auth(path, headers)` before routing;
337/// returns 401 JSON when it returns false. Keeps auth logic out of the
338/// product-agnostic core.
339pub fn handle_with_auth<F>(state: Arc<State>, req: Request, auth: F) -> std::io::Result<()>
340where
341    F: FnOnce(&str, &[Header]) -> bool,
342{
343    let path = req.url().split('?').next().unwrap_or("").to_string();
344    if !auth(&path, req.headers()) {
345        return req.respond(
346            Response::from_data(br#"{"error":"unauthorized"}"#.to_vec())
347                .with_status_code(401)
348                .with_header(
349                    Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..]).unwrap(),
350                ),
351        );
352    }
353    handle(state, req)
354}
355
356fn route(
357    state: &State,
358    method: &Method,
359    path: &str,
360    query: &str,
361    body: &str,
362    x_lex_user: Option<&str>,
363) -> Response<std::io::Cursor<Vec<u8>>> {
364    match (method, path) {
365        // ---- lex-tea v2 (HTML browser) ------------------------
366        (Method::Get, "/") => crate::web::activity_handler(state),
367        (Method::Get, "/web/branches") => crate::web::branches_handler(state),
368        (Method::Get, "/web/trust") => crate::web::trust_handler(state),
369        (Method::Get, "/web/attention") => crate::web::attention_handler(state),
370        (Method::Get, p) if p.starts_with("/web/branch/") => {
371            let name = &p["/web/branch/".len()..];
372            crate::web::branch_handler(state, name)
373        }
374        (Method::Get, p) if p.starts_with("/web/stage/") => {
375            let id = &p["/web/stage/".len()..];
376            crate::web::stage_html_handler(state, id)
377        }
378        // lex-tea v3 human-triage actions (#172). HTML forms post
379        // to /web/stage/<id>/{pin,defer,block,unblock} with a
380        // `reason` body. All four share one handler; the verb in
381        // the path picks the AttestationKind.
382        (Method::Post, p) if p.starts_with("/web/stage/") && (
383            p.ends_with("/pin") || p.ends_with("/defer")
384            || p.ends_with("/block") || p.ends_with("/unblock")
385        ) => {
386            let prefix_len = "/web/stage/".len();
387            let last_slash = p.rfind('/').unwrap_or(p.len());
388            let id = &p[prefix_len..last_slash];
389            let verb = &p[last_slash + 1..];
390            let decision = match verb {
391                "pin"     => crate::web::WebStageDecision::Pin,
392                "defer"   => crate::web::WebStageDecision::Defer,
393                "block"   => crate::web::WebStageDecision::Block,
394                "unblock" => crate::web::WebStageDecision::Unblock,
395                _ => unreachable!("matched in outer guard"),
396            };
397            crate::web::stage_decision_handler(state, id, body, decision, x_lex_user)
398        }
399        // ---- JSON API -----------------------------------------
400        (Method::Get, "/v1/health") => json_response(200, &serde_json::json!({"ok": true})),
401        (Method::Post, "/v1/parse") => parse_handler(body),
402        (Method::Post, "/v1/check") => check_handler(body),
403        (Method::Post, "/v1/publish") => publish_handler(state, body),
404        (Method::Post, "/v1/patch") => patch_handler(state, body),
405        (Method::Get, p) if p.starts_with("/v1/stage/") => {
406            let suffix = &p["/v1/stage/".len()..];
407            // Match `/v1/stage/<id>/attestations` first so a literal
408            // stage_id of "attestations" can't be misrouted.
409            if let Some(id) = suffix.strip_suffix("/attestations") {
410                stage_attestations_handler(state, id)
411            } else {
412                stage_handler(state, suffix)
413            }
414        }
415        (Method::Post, "/v1/run") => run_handler(state, body, false),
416        (Method::Post, "/v1/replay") => run_handler(state, body, true),
417        (Method::Get, p) if p.starts_with("/v1/trace/") => {
418            let id = &p["/v1/trace/".len()..];
419            trace_handler(state, id)
420        }
421        (Method::Get, "/v1/diff") => diff_handler(state, query),
422        (Method::Post, "/v1/merge/start") => merge_start_handler(state, body),
423        (Method::Post, p) if p.starts_with("/v1/merge/") && p.ends_with("/resolve") => {
424            let id = &p["/v1/merge/".len()..p.len() - "/resolve".len()];
425            merge_resolve_handler(state, id, body)
426        }
427        (Method::Post, p) if p.starts_with("/v1/merge/") && p.ends_with("/commit") => {
428            let id = &p["/v1/merge/".len()..p.len() - "/commit".len()];
429            merge_commit_handler(state, id)
430        }
431        // ---- #242: append-only sync of op log + attestation log
432        (Method::Post, "/v1/ops/batch") => ops_batch_handler(state, body),
433        (Method::Post, "/v1/attestations/batch") => attestations_batch_handler(state, body),
434        // Content half of push/pull: the stage (code) and intent blobs the
435        // op records reference. `batch` receives, `fetch` returns by id.
436        (Method::Post, "/v1/stages/batch") => crate::sync_http::stages_batch_handler(state, body),
437        (Method::Post, "/v1/stages/fetch") => crate::sync_http::stages_fetch_handler(state, body),
438        (Method::Post, "/v1/stages/missing") => crate::sync_http::stages_missing_handler(state, body),
439        (Method::Post, "/v1/intents/batch") => crate::sync_http::intents_batch_handler(state, body),
440        (Method::Post, "/v1/intents/fetch") => crate::sync_http::intents_fetch_handler(state, body),
441        // #930 P2b-1: committed lockfiles travel with the package so the
442        // write-time gate can resolve a head's pinned dependencies.
443        (Method::Post, "/v1/locks/batch") => crate::sync_http::locks_batch_handler(state, body),
444        (Method::Post, "/v1/locks/fetch") => crate::sync_http::locks_fetch_handler(state, body),
445        // #949 phase 1: typed issues travel with the package (content-
446        // addressed, like intents) so a pulled op-log carries its work items.
447        (Method::Post, "/v1/issues/batch") => crate::sync_http::issues_batch_handler(state, body),
448        (Method::Post, "/v1/issues/fetch") => crate::sync_http::issues_fetch_handler(state, body),
449        (Method::Get, "/v1/issues/list") => crate::sync_http::issues_list_handler(state),
450        // #949 phase 3: derived issue/project state, computed from the log —
451        // a board is a view, nobody drags cards. The `/list` literal above
452        // wins over the `/v1/issues/<id>` prefix arm below.
453        (Method::Get, "/v1/issues") => crate::issues_http::issues_state_handler(state),
454        (Method::Get, "/v1/projects") => crate::issues_http::projects_handler(state),
455        (Method::Get, p) if p.starts_with("/v1/issues/") => {
456            crate::issues_http::issue_detail_handler(state, &p["/v1/issues/".len()..])
457        }
458        // ---- #839 follow-up: branch management over HTTP so a remote
459        // client can create/switch branches (and thus drive the merge
460        // gates end to end), not just probe heads.
461        (Method::Get, "/v1/review/inbox") => crate::review_http::review_inbox_handler(state, query),
462        (Method::Post, "/v1/review/verdict") => crate::review_http::review_verdict_handler(state, body),
463        (Method::Get, "/v1/branches") => crate::branches_http::branches_list_handler(state),
464        (Method::Post, "/v1/branches") => crate::branches_http::branch_create_handler(state, body),
465        (Method::Post, p) if p.starts_with("/v1/branches/") && p.ends_with("/checkout") => {
466            let name = &p["/v1/branches/".len()..p.len() - "/checkout".len()];
467            crate::branches_http::branch_checkout_handler(state, name)
468        }
469        // Branch head: GET probes it (for `op push`'s delta), POST advances
470        // it (the ref half of push, fast-forward-only). Both in branches_http.
471        (Method::Get, p) if p.starts_with("/v1/branches/") && p.ends_with("/head") => {
472            let name = &p["/v1/branches/".len()..p.len() - "/head".len()];
473            crate::branches_http::branch_head_handler(state, name)
474        }
475        (Method::Post, p) if p.starts_with("/v1/branches/") && p.ends_with("/head") => {
476            let name = &p["/v1/branches/".len()..p.len() - "/head".len()];
477            crate::branches_http::branch_advance_head_handler(state, name, body)
478        }
479        // ---- #260: append-only fetch (inverse of #242 push)
480        // Body is a JSON array of OperationRecords reachable from
481        // `branch.head_op` but not from `after`, oldest-first.
482        (Method::Get, "/v1/ops/since") => ops_since_handler(state, query),
483        (Method::Get, "/v1/attestations/since") => attestations_since_handler(state, query),
484        // ---- #4: package concept ----------------------------------
485        // POST /v1/pkg/publish is handled in handle() before route()
486        // (binary body), so it doesn't appear here.
487        (Method::Get, "/v1/pkg") => pkg_list_handler(state),
488        // Owner-only visibility toggle (authed via the front door). Must
489        // precede the generic `/v1/pkg/{name}` arms; it's a PUT, so it
490        // can't collide with the GET/DELETE arms regardless.
491        (Method::Put, p) if p.starts_with("/v1/pkg/") && p.ends_with("/visibility") => {
492            let name = &p["/v1/pkg/".len()..p.len() - "/visibility".len()];
493            pkg_set_visibility_handler(state, name, body)
494        }
495        // Cut an immutable versioned release of an op-log-hosted package
496        // (#893). POST, before the generic /v1/pkg/{name} arms.
497        (Method::Post, p) if p.starts_with("/v1/pkg/") && p.ends_with("/release") => {
498            let name = &p["/v1/pkg/".len()..p.len() - "/release".len()];
499            pkg_release_handler(state, name, body)
500        }
501        (Method::Get, p) if p.starts_with("/v1/pkg/") && p.ends_with("/head") => {
502            let name = &p["/v1/pkg/".len()..p.len() - "/head".len()];
503            pkg_head_handler(state, name)
504        }
505        (Method::Get, p) if p.starts_with("/v1/pkg/") && p.ends_with("/versions") => {
506            let name = &p["/v1/pkg/".len()..p.len() - "/versions".len()];
507            pkg_versions_handler(state, name)
508        }
509        (Method::Get, p) if p.starts_with("/v1/pkg/") && p.ends_with("/api-diff") => {
510            let name = &p["/v1/pkg/".len()..p.len() - "/api-diff".len()];
511            pkg_api_diff_handler(state, name, query)
512        }
513        // /v1/pkg/{name}/{version}/archive — must match before the generic /{name}/{version}
514        (Method::Get, p) if p.starts_with("/v1/pkg/") && p.ends_with("/archive") => {
515            let inner = &p["/v1/pkg/".len()..p.len() - "/archive".len()];
516            // inner = "{name}/{version}"
517            if let Some((name, version)) = inner.split_once('/') {
518                pkg_archive_handler(state, name, version)
519            } else {
520                error_response(400, "expected /v1/pkg/{name}/{version}/archive")
521            }
522        }
523        // /v1/pkg/{name}/{version}
524        (Method::Get, p) if p.starts_with("/v1/pkg/") && p["/v1/pkg/".len()..].contains('/') => {
525            let inner = &p["/v1/pkg/".len()..];
526            if let Some((name, version)) = inner.split_once('/') {
527                pkg_get_version_handler(state, name, version)
528            } else {
529                error_response(400, "expected /v1/pkg/{name}/{version}")
530            }
531        }
532        (Method::Get, p) if p.starts_with("/v1/pkg/") => {
533            let name = &p["/v1/pkg/".len()..];
534            pkg_get_handler(state, name)
535        }
536        (Method::Delete, p) if p.starts_with("/v1/pkg/") => {
537            let name = &p["/v1/pkg/".len()..];
538            pkg_delete_handler(state, name)
539        }
540        _ => error_response(404, format!("unknown route: {method:?} {path}")),
541    }
542}
543
544#[derive(Deserialize)]
545struct ParseReq { source: String }
546
547fn parse_handler(body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
548    let req: ParseReq = match serde_json::from_str(body) {
549        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
550    };
551    match load_program_from_str(&req.source) {
552        Ok(prog) => {
553            let stages = canonicalize_program(&prog);
554            json_response(200, &serde_json::to_value(&stages).unwrap())
555        }
556        Err(e) => error_response(400, format!("syntax error: {e}")),
557    }
558}
559
560pub(crate) fn check_handler(body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
561    let req: ParseReq = match serde_json::from_str(body) {
562        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
563    };
564    let prog = match load_program_from_str(&req.source) {
565        Ok(p) => p, Err(e) => return error_response(400, format!("syntax error: {e}")),
566    };
567    let stages = canonicalize_program(&prog);
568    match lex_types::check_program(&stages) {
569        Ok(_) => json_response(200, &serde_json::json!({"ok": true})),
570        Err(errs) => json_response(422, &serde_json::to_value(&errs).unwrap()),
571    }
572}
573
574#[derive(Deserialize)]
575struct PublishReq { source: String, #[serde(default)] activate: bool }
576
577pub(crate) fn publish_handler(state: &State, body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
578    let req: PublishReq = match serde_json::from_str(body) {
579        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
580    };
581    let prog = match load_program_from_str(&req.source) {
582        Ok(p) => p, Err(e) => return error_response(400, format!("syntax error: {e}")),
583    };
584    // #168: rewrite stdlib parse calls to parse_strict so the
585    // bytecode emitted from these stages enforces required-field
586    // checks at runtime.
587    let mut stages = canonicalize_program(&prog);
588    if let Err(errs) = lex_types::check_and_rewrite_program(&mut stages) {
589        return error_with_detail(422, "type errors", serde_json::to_value(&errs).unwrap());
590    }
591    // #835 Tier 1: behavioral example gate. check_and_rewrite_program
592    // only type-checks `examples {}`; run them and refuse the publish
593    // if any declared example evaluates to the wrong value. Non-breaking:
594    // functions without examples (and effectful ones, which can't have
595    // them) produce no cases.
596    let example_errors = lex_runtime::evaluate_examples(&stages);
597    if !example_errors.is_empty() {
598        return error_with_detail(422, "example mismatch",
599            serde_json::to_value(&example_errors).unwrap_or_default());
600    }
601
602    let store = state.store.lock().unwrap();
603    let branch = store.current_branch();
604
605    // Compute diff between what's already on the branch and the new program.
606    let old_head = match store.branch_head(&branch) {
607        Ok(h) => h,
608        Err(e) => return error_response(500, format!("branch_head: {e}")),
609    };
610    // Fns + types (#895) on both sides. Old side is the branch head.
611    let old_head_stages: Vec<lex_ast::Stage> =
612        old_head.values().filter_map(|stg| store.get_ast(stg).ok()).collect();
613    let old_fns = stage_fns(&old_head_stages);
614    let new_fns = stage_fns(&stages);
615    let old_types = stage_types(&old_head_stages);
616    let new_types = stage_types(&stages);
617    let report =
618        lex_vcs::compute_diff_with_types(&old_fns, &new_fns, &old_types, &new_types, false);
619
620    // Build new imports map from any Import stages in the source.
621    let mut new_imports: lex_vcs::ImportMap = lex_vcs::ImportMap::new();
622    {
623        let entry = new_imports.entry("<source>".into()).or_default();
624        for s in &stages {
625            if let lex_ast::Stage::Import(im) = s {
626                entry.insert(lex_vcs::ImportRef {
627                    reference: im.reference.clone(),
628                    alias: im.alias.clone(),
629                });
630            }
631        }
632    }
633
634    match store.publish_program(&branch, &stages, &report, &new_imports, req.activate) {
635        Ok(outcome) => {
636            // #835 Tier 1: record the behavioral-examples verdict for each
637            // published fn-stage that declares examples. Best-effort — a
638            // failure to record must not fail an otherwise-good publish.
639            record_examples_for_publish(&store, &stages, &outcome);
640            json_response(200, &serde_json::json!({
641                "ops": outcome.ops,
642                "head_op": outcome.head_op,
643            }))
644        }
645        // The store-write gate (#130) also type-checks at the top
646        // of `publish_program`. The handler above already pre-checks,
647        // so this branch is reached only on a race or a state we
648        // didn't see at handler time. Surface the structured
649        // envelope (422) instead of a generic 500 — same shape the
650        // initial pre-check uses, so a client only has one error
651        // contract to handle.
652        Err(lex_store::StoreError::TypeError(errs)) => {
653            error_with_detail(422, "type errors", serde_json::to_value(&errs).unwrap())
654        }
655        Err(e) => write_error_response("publish_program", e),
656    }
657}
658
659#[derive(Deserialize)]
660struct PatchReq {
661    stage_id: String,
662    patch: lex_ast::Patch,
663    #[serde(default)] activate: bool,
664}
665
666/// POST /v1/patch — apply a structured edit to a stored stage's
667/// canonical AST, type-check the result, and publish a new stage.
668fn patch_handler(state: &State, body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
669    let req: PatchReq = match serde_json::from_str(body) {
670        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
671    };
672    let store = state.store.lock().unwrap();
673
674    // 1. Load.
675    let original = match store.get_ast(&req.stage_id) {
676        Ok(s) => s, Err(e) => return error_response(404, format!("stage: {e}")),
677    };
678
679    // 2. Apply.
680    let patched = match lex_ast::apply_patch(&original, &req.patch) {
681        Ok(s) => s,
682        Err(e) => return error_with_detail(422, "patch failed",
683            serde_json::to_value(&e).unwrap_or_default()),
684    };
685
686    // 3. No isolated check (#833): the gated apply below type-checks
687    // the *composed* program — the branch head with the patched stage
688    // swapped in. Stricter where it matters (a body that no longer
689    // composes with its callers is refused) and correct where the old
690    // isolated check was wrong (a body calling a sibling was rejected
691    // as an unknown identifier a one-stage program couldn't see).
692
693    // Routing through the gated apply so /v1/patch participates in the
694    // op DAG. We know this op is always a body change on the existing
695    // sig (a patch can't add a brand-new fn).
696    let branch = store.current_branch();
697
698    // Find the sig — patched stage's sig must match the original's.
699    let sig = match lex_ast::sig_id(&patched) {
700        Some(s) => s,
701        None => return error_response(500, "patched stage has no sig_id"),
702    };
703
704    // Persist before the gate (its RepairHint on rejection is
705    // addressed to this stage); activate only once the head moved.
706    let new_id = match store.publish(&patched) {
707        Ok(id) => id, Err(e) => return error_response(500, format!("publish: {e}")),
708    };
709
710    // Determine op kind: ChangeEffectSig if effects differ, ModifyBody otherwise.
711    let original_effects: std::collections::BTreeSet<String> = match &original {
712        lex_ast::Stage::FnDecl(fd) => fd.effects.iter().map(|e| e.name.clone()).collect(),
713        _ => std::collections::BTreeSet::new(),
714    };
715    let patched_effects: std::collections::BTreeSet<String> = match &patched {
716        lex_ast::Stage::FnDecl(fd) => fd.effects.iter().map(|e| e.name.clone()).collect(),
717        _ => std::collections::BTreeSet::new(),
718    };
719    let head_now = match store.get_branch(&branch) {
720        Ok(b) => b.and_then(|b| b.head_op),
721        Err(e) => return error_response(500, format!("get_branch: {e}")),
722    };
723    let kind = if original_effects != patched_effects {
724        // #247: budget delta is part of the canonical payload now.
725        // Patch endpoints don't currently rehydrate the AST to
726        // recompute budgets, so leave them None — clients that
727        // need budget tracking should publish through the diff
728        // pipeline (`lex publish`) where `compute_diff` populates
729        // them.
730        let from_budget = lex_vcs::operation_budget_from_effects(&original_effects);
731        let to_budget = lex_vcs::operation_budget_from_effects(&patched_effects);
732        // #992: the patched stage declares the new effects, so it hashes to a
733        // different SigId and the store files it under that one. Record it, or
734        // the head keeps the old sig pointing at a stage no store holds there.
735        let to_sig_id = lex_ast::sig_id(&patched).filter(|s| *s != sig);
736        lex_vcs::OperationKind::ChangeEffectSig {
737            sig_id: sig.clone(),
738            from_stage_id: req.stage_id.clone(),
739            to_stage_id: new_id.clone(),
740            from_effects: original_effects,
741            to_effects: patched_effects,
742            from_budget,
743            to_budget,
744            to_sig_id,
745        }
746    } else {
747        let budget = lex_vcs::operation_budget_from_effects(&original_effects);
748        lex_vcs::OperationKind::ModifyBody {
749            sig_id: sig.clone(),
750            from_stage_id: req.stage_id.clone(),
751            to_stage_id: new_id.clone(),
752            from_budget: budget,
753            to_budget: budget,
754            // #992: a patch that touches the signature (types, examples)
755            // moves the sig just as an effect change does.
756            to_sig_id: lex_ast::sig_id(&patched).filter(|s| *s != sig),
757        }
758    };
759    // #992: derive the transition from the op, never hard-code `Replace`. A
760    // sig-moving op must retire the old sig and bind the new one; a `Replace`
761    // here left `(old_sig, new_stage)` at the head — a pair no store holds.
762    let transition = lex_store::transition_for_kind(&kind);
763    let op = lex_vcs::Operation::new(
764        kind,
765        head_now.into_iter().collect::<Vec<_>>(),
766    );
767    let op_id = match store.apply_operation_gated(&branch, op, transition) {
768        Ok(id) => id,
769        Err(lex_store::StoreError::TypeError(errs)) => return error_with_detail(
770            422, "type errors after patch", serde_json::to_value(&errs).unwrap_or_default()),
771        Err(e) => return write_error_response("apply_operation_gated", e),
772    };
773    if req.activate {
774        if let Err(e) = store.activate(&new_id) {
775            return error_response(500, format!("activate: {e}"));
776        }
777    }
778
779    let status = format!("{:?}",
780        store.get_status(&new_id).unwrap_or(lex_store::StageStatus::Draft)).to_lowercase();
781    json_response(200, &serde_json::json!({
782        "old_stage_id": req.stage_id,
783        "new_stage_id": new_id,
784        "sig_id": sig,
785        "status": status,
786        "op_id": op_id,
787    }))
788}
789
790pub(crate) fn stage_handler(state: &State, id: &str) -> Response<std::io::Cursor<Vec<u8>>> {
791    let store = state.store.lock().unwrap();
792    let meta = match store.get_metadata(id) {
793        Ok(m) => m, Err(e) => return error_response(404, format!("{e}")),
794    };
795    let ast = match store.get_ast(id) {
796        Ok(a) => a, Err(e) => return error_response(404, format!("{e}")),
797    };
798    let status = format!("{:?}", store.get_status(id).unwrap_or(lex_store::StageStatus::Draft)).to_lowercase();
799    json_response(200, &serde_json::json!({
800        "metadata": meta,
801        "ast": ast,
802        "status": status,
803    }))
804}
805
806/// `GET /v1/stage/<id>/attestations` — every persisted attestation
807/// for this stage, newest-first by timestamp. Issue #132's
808/// queryable-evidence consumer surface.
809///
810/// 404s on unknown stage_id (matches `/v1/stage/<id>`'s shape so a
811/// caller round-tripping both endpoints sees consistent errors).
812/// Empty list (200) is *evidence of absence*: the stage exists but
813/// no producer has attested it.
814pub(crate) fn stage_attestations_handler(state: &State, id: &str) -> Response<std::io::Cursor<Vec<u8>>> {
815    let store = state.store.lock().unwrap();
816    if let Err(e) = store.get_metadata(id) {
817        return error_response(404, format!("{e}"));
818    }
819    let log = match store.attestation_log() {
820        Ok(l) => l,
821        Err(e) => return error_response(500, format!("attestation log: {e}")),
822    };
823    let mut listing = match log.list_for_stage(&id.to_string()) {
824        Ok(v) => v,
825        Err(e) => return error_response(500, format!("list_for_stage: {e}")),
826    };
827    listing.sort_by_key(|a| std::cmp::Reverse(a.timestamp));
828    json_response(200, &serde_json::json!({"attestations": listing}))
829}
830
831#[derive(Deserialize, Default)]
832struct PolicyJson {
833    #[serde(default)] allow_effects: Vec<String>,
834    #[serde(default)] allow_fs_read: Vec<String>,
835    #[serde(default)] allow_fs_write: Vec<String>,
836    #[serde(default)] budget: Option<u64>,
837}
838
839impl PolicyJson {
840    fn into_policy(self) -> Policy {
841        Policy {
842            allow_effects: self.allow_effects.into_iter().collect::<BTreeSet<_>>(),
843            allow_fs_read: self.allow_fs_read.into_iter().map(PathBuf::from).collect(),
844            allow_fs_write: self.allow_fs_write.into_iter().map(PathBuf::from).collect(),
845            allow_net_host: Vec::new(),
846            allow_proc: Vec::new(),
847            allow_approval: Vec::new(),
848            budget: self.budget,
849        }
850    }
851}
852
853#[derive(Deserialize)]
854struct RunReq {
855    source: String,
856    #[serde(rename = "fn")] func: String,
857    #[serde(default)] args: Vec<serde_json::Value>,
858    #[serde(default)] policy: PolicyJson,
859    #[serde(default)] overrides: IndexMap<String, serde_json::Value>,
860}
861
862pub(crate) fn run_handler(state: &State, body: &str, with_overrides: bool) -> Response<std::io::Cursor<Vec<u8>>> {
863    let req: RunReq = match serde_json::from_str(body) {
864        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
865    };
866    let prog = match load_program_from_str(&req.source) {
867        Ok(p) => p, Err(e) => return error_response(400, format!("syntax error: {e}")),
868    };
869    let stages = canonicalize_program(&prog);
870    if let Err(errs) = lex_types::check_program(&stages) {
871        return error_with_detail(422, "type errors", serde_json::to_value(&errs).unwrap());
872    }
873    let bc = compile_program(&stages);
874    let mut policy = req.policy.into_policy();
875    // When a server-imposed ceiling is present (multi-tenant
876    // embedders like lex-hub), the request policy can only narrow
877    // it — never grant itself proc/fs/net beyond what the operator
878    // allowed. Single-tenant `lex serve` leaves the ceiling unset
879    // and runs the caller's policy verbatim.
880    if let Some(ceiling) = &state.policy_ceiling {
881        policy = clamp_policy(policy, ceiling);
882    }
883    if let Err(violations) = check_policy(&bc, &policy) {
884        return error_with_detail(403, "policy violation", serde_json::to_value(&violations).unwrap());
885    }
886
887    let mut recorder = lex_trace::Recorder::new();
888    if with_overrides && !req.overrides.is_empty() {
889        recorder = recorder.with_overrides(req.overrides);
890    }
891    let handle = recorder.handle();
892    let handler = DefaultHandler::new(policy);
893    let mut vm = Vm::with_handler(&bc, Box::new(handler));
894    vm.set_tracer(Box::new(recorder));
895
896    let vargs: Vec<Value> = req.args.iter().map(json_to_value).collect();
897    let started = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs();
898    let result = vm.call(&req.func, vargs);
899    let ended = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs();
900
901    let store = state.store.lock().unwrap();
902    let (root_out, root_err, status) = match &result {
903        Ok(v) => (Some(value_to_json(v)), None, 200u16),
904        Err(e) => (None, Some(format!("{e}")), 200u16),
905    };
906    let tree = handle.finalize(req.func.clone(), serde_json::Value::Null,
907        root_out.clone(), root_err.clone(), started, ended);
908    let run_id = match store.save_trace(&tree) {
909        Ok(id) => id,
910        Err(e) => return error_response(500, format!("save_trace: {e}")),
911    };
912
913    let mut body = serde_json::json!({
914        "run_id": run_id,
915        "output": root_out,
916    });
917    if let Some(err) = root_err {
918        body["error"] = serde_json::Value::String(err);
919    }
920    json_response(status, &body)
921}
922
923fn trace_handler(state: &State, id: &str) -> Response<std::io::Cursor<Vec<u8>>> {
924    let store = state.store.lock().unwrap();
925    match store.load_trace(id) {
926        Ok(t) => json_response(200, &serde_json::to_value(&t).unwrap()),
927        Err(e) => error_response(404, format!("{e}")),
928    }
929}
930
931fn diff_handler(state: &State, query: &str) -> Response<std::io::Cursor<Vec<u8>>> {
932    let mut a = None;
933    let mut b = None;
934    for kv in query.split('&') {
935        if let Some((k, v)) = kv.split_once('=') {
936            match k { "a" => a = Some(v.to_string()), "b" => b = Some(v.to_string()), _ => {} }
937        }
938    }
939    let (Some(a), Some(b)) = (a, b) else {
940        return error_response(400, "missing a or b query params");
941    };
942    let store = state.store.lock().unwrap();
943    let ta = match store.load_trace(&a) { Ok(t) => t, Err(e) => return error_response(404, format!("a: {e}")) };
944    let tb = match store.load_trace(&b) { Ok(t) => t, Err(e) => return error_response(404, format!("b: {e}")) };
945    match lex_trace::diff_runs(&ta, &tb) {
946        Some(d) => json_response(200, &serde_json::to_value(&d).unwrap()),
947        None => json_response(200, &serde_json::json!({"divergence": null})),
948    }
949}
950
951fn json_to_value(v: &serde_json::Value) -> Value { Value::from_json(v) }
952
953fn value_to_json(v: &Value) -> serde_json::Value { v.to_json() }
954
955#[derive(Deserialize)]
956struct MergeStartReq {
957    src_branch: String,
958    dst_branch: String,
959}
960
961/// `POST /v1/merge/start` (#134) — open a stateful merge between two
962/// branch heads and return the conflicts the agent needs to
963/// resolve. Auto-resolved sigs (one-sided changes, identical
964/// changes both sides) are returned for audit but don't block
965/// commit.
966///
967/// Response: `{ merge_id, src_head, dst_head, lca, conflicts,
968/// auto_resolved_count }`. The session is held in process memory
969/// keyed by `merge_id` for subsequent `resolve` / `commit` calls
970/// (next slices).
971fn merge_start_handler(state: &State, body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
972    let req: MergeStartReq = match serde_json::from_str(body) {
973        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
974    };
975    let store = state.store.lock().unwrap();
976    let src_head = match store.get_branch(&req.src_branch) {
977        Ok(Some(b)) => b.head_op,
978        Ok(None) => return error_response(404, format!("unknown src branch `{}`", req.src_branch)),
979        Err(e) => return error_response(500, format!("src branch read: {e}")),
980    };
981    let dst_head = match store.get_branch(&req.dst_branch) {
982        Ok(Some(b)) => b.head_op,
983        Ok(None) => return error_response(404, format!("unknown dst branch `{}`", req.dst_branch)),
984        Err(e) => return error_response(500, format!("dst branch read: {e}")),
985    };
986    let log = match lex_vcs::OpLog::open(store.root()) {
987        Ok(l) => l,
988        Err(e) => return error_response(500, format!("op log: {e}")),
989    };
990    // Caller doesn't choose merge_ids — minted server-side from
991    // wall clock + a per-process counter avoids leaking session
992    // ids' shape into the public surface.
993    let merge_id = mint_merge_id();
994    let session = match MergeSession::start(
995        merge_id.clone(),
996        &log,
997        src_head.as_ref(),
998        dst_head.as_ref(),
999    ) {
1000        Ok(s) => s,
1001        Err(e) => return error_response(500, format!("merge start: {e}")),
1002    };
1003    let conflicts: Vec<&lex_vcs::ConflictRecord> = session.remaining_conflicts();
1004    let auto_resolved_count = session.auto_resolved.len();
1005    let body = serde_json::json!({
1006        "merge_id": merge_id,
1007        "src_head": session.src_head,
1008        "dst_head": session.dst_head,
1009        "lca":      session.lca,
1010        "conflicts": conflicts,
1011        "auto_resolved_count": auto_resolved_count,
1012    });
1013    drop(conflicts);
1014    drop(store);
1015    let wrapped = ApiMergeSession {
1016        inner: session,
1017        src_branch: req.src_branch,
1018        dst_branch: req.dst_branch,
1019    };
1020    state.sessions.lock().unwrap().insert(merge_id, wrapped);
1021    json_response(200, &body)
1022}
1023
1024#[derive(Deserialize)]
1025struct MergeResolveReq {
1026    /// Each entry is `(conflict_id, resolution)`. The resolution is
1027    /// the same shape as `lex_vcs::Resolution`'s tagged JSON form
1028    /// — `{"kind":"take_ours"}`, `{"kind":"take_theirs"}`,
1029    /// `{"kind":"defer"}`, or `{"kind":"custom","op":{...}}`.
1030    resolutions: Vec<MergeResolveEntry>,
1031}
1032
1033#[derive(Deserialize)]
1034struct MergeResolveEntry {
1035    conflict_id: String,
1036    resolution: lex_vcs::Resolution,
1037}
1038
1039/// `POST /v1/merge/<id>/resolve` (#134) — submit batched
1040/// resolutions against the conflicts surfaced by `merge/start`.
1041/// Returns one verdict per input: accepted (recorded against the
1042/// session) or rejected (with structured reason). The session
1043/// stays alive across calls so an agent can iterate.
1044///
1045/// Errors:
1046/// - 404 if `merge_id` doesn't refer to a live session (a typo
1047///   or a session GC'd by a server restart).
1048/// - 400 on malformed body.
1049fn merge_resolve_handler(
1050    state: &State,
1051    merge_id: &str,
1052    body: &str,
1053) -> Response<std::io::Cursor<Vec<u8>>> {
1054    let req: MergeResolveReq = match serde_json::from_str(body) {
1055        Ok(r) => r, Err(e) => return error_response(400, format!("bad request: {e}")),
1056    };
1057    let mut sessions = state.sessions.lock().unwrap();
1058    let Some(wrapped) = sessions.get_mut(merge_id) else {
1059        return error_response(404, format!("unknown merge_id `{merge_id}`"));
1060    };
1061    let pairs: Vec<(String, lex_vcs::Resolution)> = req.resolutions.into_iter()
1062        .map(|e| (e.conflict_id, e.resolution))
1063        .collect();
1064    // #834: type-check each resolution against dst's head at submission
1065    // time (not only at commit) so the agent's "submit N, see which
1066    // broke, retry" loop gets its feedback here. The store composes the
1067    // projected program; the session owns the merge→delta semantics.
1068    let store = state.store.lock().unwrap();
1069    let checker = lex_store::MergeResolutionChecker::new(&store, wrapped.dst_branch.clone());
1070    let verdicts = wrapped.inner.resolve_checked(pairs, &checker);
1071    drop(store);
1072    let remaining: Vec<&lex_vcs::ConflictRecord> = wrapped.inner.remaining_conflicts();
1073    let body = serde_json::json!({
1074        "verdicts": verdicts,
1075        "remaining_conflicts": remaining,
1076    });
1077    json_response(200, &body)
1078}
1079
1080/// `POST /v1/merge/<id>/commit` (#134) — finalize a merge
1081/// session. Builds a `Merge` op from the auto-resolved sigs +
1082/// the conflict resolutions, applies it to the dst branch, and
1083/// returns the new head op id. The session is dropped on
1084/// success; the caller would re-run `merge/start` to land
1085/// further changes.
1086///
1087/// Errors:
1088/// - 404: unknown `merge_id`.
1089/// - 422: conflicts remaining (pass `Defer` or just don't
1090///   resolve a conflict and you land here). Body carries the
1091///   list so the caller knows which still need attention.
1092/// - 422: a `Custom` resolution was used. The data layer
1093///   supports them but landing them via HTTP needs an extra
1094///   pass to apply the custom op against the dst branch
1095///   first; deferred to a follow-up slice. Use TakeOurs /
1096///   TakeTheirs for now.
1097/// - 409: dependency conflict (#977) — the branches pin the same
1098///   package at different versions, so no merged lock exists. Body
1099///   `detail` names the package and both versions; nothing is written.
1100/// - 500: filesystem error while landing the merge op.
1101fn merge_commit_handler(
1102    state: &State,
1103    merge_id: &str,
1104) -> Response<std::io::Cursor<Vec<u8>>> {
1105    use std::collections::BTreeMap;
1106    let wrapped = match state.sessions.lock().unwrap().remove(merge_id) {
1107        Some(w) => w,
1108        None => return error_response(404, format!("unknown merge_id `{merge_id}`")),
1109    };
1110    let dst_branch = wrapped.dst_branch.clone();
1111    let src_head = wrapped.inner.src_head.clone();
1112    let dst_head = wrapped.inner.dst_head.clone();
1113    let auto_resolved = wrapped.inner.auto_resolved.clone();
1114
1115    // Translate auto-resolved + resolutions into the StageTransition::Merge
1116    // entries map. Only sigs whose head changes relative to dst go in.
1117    let mut entries: BTreeMap<lex_vcs::SigId, Option<lex_vcs::StageId>> = BTreeMap::new();
1118
1119    // Auto-resolved: only `Src` (one-sided change on src) modifies dst.
1120    for outcome in &auto_resolved {
1121        if let lex_vcs::MergeOutcome::Src { sig_id, stage_id } = outcome {
1122            entries.insert(sig_id.clone(), stage_id.clone());
1123        }
1124    }
1125
1126    // Conflict resolutions.
1127    let resolved = match wrapped.inner.commit() {
1128        Ok(r) => r,
1129        Err(lex_vcs::CommitError::ConflictsRemaining(ids)) => {
1130            // Re-insert isn't possible since we removed above; the
1131            // caller will need to re-start. That's acceptable: a
1132            // commit-with-unresolved-conflicts is operator error.
1133            return error_with_detail(
1134                422,
1135                "conflicts remaining",
1136                serde_json::json!({"unresolved": ids}),
1137            );
1138        }
1139    };
1140
1141    for (conflict_id, resolution) in resolved {
1142        match resolution {
1143            lex_vcs::Resolution::TakeOurs => {
1144                // Dst already has its head. No entry needed.
1145            }
1146            lex_vcs::Resolution::TakeTheirs => {
1147                // Find the conflict's `theirs` stage_id in the
1148                // session snapshot. We don't have direct access to
1149                // it post-commit (commit consumed the session); but
1150                // we can reconstruct from `auto_resolved` plus the
1151                // session's pre-commit conflict map. Since we
1152                // already moved the inner session, the cleanest fix
1153                // for this slice is to rebuild from the on-disk
1154                // graph: walk src_head, find the latest stage for
1155                // the conflict's sig.
1156                match resolve_take_theirs(state, &src_head, &conflict_id) {
1157                    Ok(stage_id) => {
1158                        entries.insert(conflict_id.clone(), stage_id);
1159                    }
1160                    Err(e) => return error_response(500, format!("resolve take_theirs: {e}")),
1161                }
1162            }
1163            lex_vcs::Resolution::Custom { op } => {
1164                // The agent's brand-new op carries the merge target
1165                // in its kind (e.g. ModifyBody.to_stage_id). The op
1166                // itself isn't separately recorded in the log here
1167                // — its head-map effect is folded into the merge
1168                // op's entries map. Callers that want the op as a
1169                // first-class history entry should publish it via
1170                // /v1/publish first and submit a TakeTheirs/TakeOurs
1171                // resolution against the resulting head.
1172                match op.kind.merge_target() {
1173                    Some((sig, stage)) => {
1174                        if sig != conflict_id {
1175                            return error_with_detail(
1176                                422,
1177                                "custom op targets a different sig than the conflict",
1178                                serde_json::json!({
1179                                    "conflict_id": conflict_id,
1180                                    "op_targets": sig,
1181                                }),
1182                            );
1183                        }
1184                        entries.insert(conflict_id, stage);
1185                    }
1186                    None => {
1187                        return error_with_detail(
1188                            422,
1189                            "custom op kind doesn't yield a single sig→stage delta",
1190                            serde_json::json!({
1191                                "conflict_id": conflict_id,
1192                                "kind": serde_json::to_value(&op.kind).unwrap_or(serde_json::Value::Null),
1193                            }),
1194                        );
1195                    }
1196                }
1197            }
1198            lex_vcs::Resolution::Defer => {
1199                // Unreachable: commit() rejects Defer above.
1200                return error_response(500, "internal: Defer slipped past commit gate");
1201            }
1202        }
1203    }
1204
1205    let resolved_count = entries.len();
1206    let mut parents: Vec<lex_vcs::OpId> = Vec::new();
1207    if let Some(d) = dst_head { parents.push(d); }
1208    if let Some(s) = src_head { parents.push(s); }
1209    let op = lex_vcs::Operation::new(
1210        lex_vcs::OperationKind::Merge { resolved: resolved_count },
1211        parents,
1212    );
1213    let transition = lex_vcs::StageTransition::Merge { entries };
1214    let store = state.store.lock().unwrap();
1215    // Gated (#833): lands the merge op, type-checks the real
1216    // post-merge head, rolls the head back on a TypeError.
1217    match store.apply_merge_op_gated(&dst_branch, op, transition) {
1218        Ok(new_head_op) => json_response(200, &serde_json::json!({
1219            "new_head_op": new_head_op,
1220            "dst_branch": dst_branch,
1221        })),
1222        Err(lex_store::StoreError::TypeError(errs)) => error_with_detail(
1223            422, "merged program has type errors", serde_json::to_value(&errs).unwrap_or_default()),
1224        // #977: the two branches pin the same dependency at different
1225        // versions. The merge commits the union of both parents' locks and
1226        // refuses to pick a side silently; nothing was written, so the dst
1227        // branch is unchanged. 409: the request is fine, the branches'
1228        // states conflict — align the version on one branch and retry.
1229        Err(e @ lex_store::StoreError::DependencyConflict { .. }) => {
1230            let detail = match &e {
1231                lex_store::StoreError::DependencyConflict { package, dst_version, src_version } => {
1232                    serde_json::json!({
1233                        "kind": "dependency_conflict",
1234                        "package": package,
1235                        "dst_branch": dst_branch,
1236                        "dst_version": dst_version,
1237                        "src_branch": wrapped.src_branch,
1238                        "src_version": src_version,
1239                    })
1240                }
1241                _ => serde_json::Value::Null,
1242            };
1243            error_with_detail(409, e.to_string(), detail)
1244        }
1245        Err(e) => write_error_response("apply merge op", e),
1246    }
1247}
1248
1249/// Walk the op log from `src_head` backwards to find the latest
1250/// stage assigned to `sig`. Used by the commit handler to figure
1251/// out what stage `TakeTheirs` should land. `Ok(None)` means src
1252/// removed the sig.
1253fn resolve_take_theirs(
1254    state: &State,
1255    src_head: &Option<lex_vcs::OpId>,
1256    sig: &lex_vcs::SigId,
1257) -> std::io::Result<Option<lex_vcs::StageId>> {
1258    let store = state.store.lock().unwrap();
1259    let log = lex_vcs::OpLog::open(store.root())?;
1260    let Some(head) = src_head.as_ref() else { return Ok(None); };
1261    // Walk forward from root → head, replaying each op's transition
1262    // for `sig`; the last assignment wins.
1263    let mut current: Option<lex_vcs::StageId> = None;
1264    for record in log.walk_forward(head, None)? {
1265        match &record.produces {
1266            lex_vcs::StageTransition::Create { sig_id, stage_id }
1267                if sig_id == sig => { current = Some(stage_id.clone()); }
1268            lex_vcs::StageTransition::Replace { sig_id, to, .. }
1269                if sig_id == sig => { current = Some(to.clone()); }
1270            lex_vcs::StageTransition::Remove { sig_id, .. }
1271                if sig_id == sig => { current = None; }
1272            lex_vcs::StageTransition::Rename { from, to, body_stage_id }
1273                if from == sig || to == sig => {
1274                if from == sig { current = None; }
1275                if to == sig   { current = Some(body_stage_id.clone()); }
1276            }
1277            lex_vcs::StageTransition::Merge { entries } => {
1278                if let Some(opt) = entries.get(sig) {
1279                    current = opt.clone();
1280                }
1281            }
1282            _ => {}
1283        }
1284    }
1285    Ok(current)
1286}
1287
1288fn mint_merge_id() -> MergeSessionId {
1289    use std::sync::atomic::{AtomicU64, Ordering};
1290    static COUNTER: AtomicU64 = AtomicU64::new(0);
1291    let nanos = SystemTime::now()
1292        .duration_since(UNIX_EPOCH)
1293        .map(|d| d.as_nanos())
1294        .unwrap_or(0);
1295    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
1296    format!("merge_{nanos:x}_{n:x}")
1297}
1298
1299// ---- #242: append-only sync ---------------------------------------
1300
1301/// `POST /v1/ops/batch` (#242). Server endpoint for `lex op push`.
1302///
1303/// Body: a JSON array of `OperationRecord`s. The handler validates
1304/// DAG integrity by checking that every op's `parents` either
1305/// already exist on the remote *or* appear earlier in the same
1306/// batch. This lets a client send a topologically-ordered slice
1307/// without first probing for what's already there.
1308///
1309/// Response shape:
1310///
1311/// ```json
1312/// { "received": N, "added": M, "skipped": (N-M), "added_ids": [...] }
1313/// ```
1314///
1315/// Failure modes:
1316///
1317/// * `400` — body isn't a JSON array of op records.
1318/// * `422` with `{ "error": "MissingParent", "detail": { "op_id":
1319///   ..., "missing_parent": ... } }` if a parent is unreachable.
1320///   The whole batch is rejected; nothing is persisted. The client
1321///   should backfill the missing op and retry.
1322/// * `409` if the supplied `op_id` doesn't match the canonical
1323///   hash of the record's payload — content addressing must hold
1324///   over the wire.
1325///
1326/// Idempotency: a record whose `op_id` already exists is silently
1327/// skipped (not added, not rejected). Pushing the same payload
1328/// twice is `received == N, added == 0` on the second call.
1329pub(crate) fn ops_batch_handler(state: &State, body: &str)
1330    -> Response<std::io::Cursor<Vec<u8>>>
1331{
1332    let records: Vec<lex_vcs::OperationRecord> = match serde_json::from_str(body) {
1333        Ok(r) => r,
1334        Err(e) => return error_response(400,
1335            format!("body must be a JSON array of OperationRecord: {e}")),
1336    };
1337    let store = state.store.lock().unwrap();
1338    let log = match lex_vcs::OpLog::open(store.root()) {
1339        Ok(l) => l,
1340        Err(e) => return error_response(500, format!("opening op log: {e}")),
1341    };
1342
1343    // Validate every record before persisting any of them.
1344    //
1345    // 1. Content-addressing: the supplied `op_id` must match the
1346    //    canonical hash of `record.op`. Otherwise the client is
1347    //    sending a forged or corrupted record.
1348    // 2. DAG integrity: every parent must either already exist in
1349    //    the local log OR appear earlier in this batch.
1350    let mut batch_ids: std::collections::BTreeSet<lex_vcs::OpId> =
1351        std::collections::BTreeSet::new();
1352    for rec in &records {
1353        let expected = rec.op.op_id();
1354        if expected != rec.op_id {
1355            return error_with_detail(409, "OpIdMismatch", serde_json::json!({
1356                "supplied": rec.op_id,
1357                "expected": expected,
1358            }));
1359        }
1360        for parent in &rec.op.parents {
1361            let known = match log.get(parent) {
1362                Ok(Some(_)) => true,
1363                Ok(None) => false,
1364                Err(e) => return error_response(500, format!("op log read: {e}")),
1365            };
1366            if !known && !batch_ids.contains(parent) {
1367                return error_with_detail(422, "MissingParent", serde_json::json!({
1368                    "op_id": rec.op_id,
1369                    "missing_parent": parent,
1370                }));
1371            }
1372        }
1373        batch_ids.insert(rec.op_id.clone());
1374    }
1375
1376    // Persist. `OpLog::put` is idempotent so a re-push is a no-op
1377    // for already-present records.
1378    let mut added = 0usize;
1379    let mut added_ids: Vec<&lex_vcs::OpId> = Vec::new();
1380    for rec in &records {
1381        let already_present = matches!(log.get(&rec.op_id), Ok(Some(_)));
1382        match log.put(rec) {
1383            Ok(()) => {
1384                if !already_present {
1385                    added += 1;
1386                    added_ids.push(&rec.op_id);
1387                }
1388            }
1389            Err(e) => return error_response(500, format!("op log write: {e}")),
1390        }
1391    }
1392
1393    json_response(200, &serde_json::json!({
1394        "received": records.len(),
1395        "added": added,
1396        "skipped": records.len() - added,
1397        "added_ids": added_ids,
1398    }))
1399}
1400
1401/// `POST /v1/attestations/batch` (#242). Server endpoint for `lex
1402/// attest push`.
1403///
1404/// Body: a JSON array of `Attestation`s. Validates that each
1405/// attestation's `op_id` (when set) refers to an op that already
1406/// exists on the remote — `attestation_id` is then re-derivable
1407/// from the canonical form, so cross-store dedup just works.
1408///
1409/// Response: same shape as `ops_batch_handler` but `added_ids` is
1410/// the list of accepted `attestation_id`s.
1411///
1412/// Failure modes:
1413///
1414/// * `400` for malformed JSON.
1415/// * `422` with `{ "error": "UnknownOp", "detail": { ... } }` if
1416///   an attestation's `op_id` references an op the remote doesn't
1417///   know about. Whole batch rejected.
1418/// * `409` `AttestationIdMismatch` if the supplied id doesn't
1419///   match the canonical hash.
1420///
1421/// Idempotency: same as the ops endpoint — content-addressed dedup.
1422pub(crate) fn attestations_batch_handler(state: &State, body: &str)
1423    -> Response<std::io::Cursor<Vec<u8>>>
1424{
1425    let attestations: Vec<lex_vcs::Attestation> = match serde_json::from_str(body) {
1426        Ok(a) => a,
1427        Err(e) => return error_response(400,
1428            format!("body must be a JSON array of Attestation: {e}")),
1429    };
1430    let store = state.store.lock().unwrap();
1431    let log = match store.attestation_log() {
1432        Ok(l) => l,
1433        Err(e) => return error_response(500, format!("opening attestation log: {e}")),
1434    };
1435    let op_log = match lex_vcs::OpLog::open(store.root()) {
1436        Ok(l) => l,
1437        Err(e) => return error_response(500, format!("opening op log: {e}")),
1438    };
1439
1440    // Validate before persisting any record.
1441    for att in &attestations {
1442        // Content-addressing: re-derive attestation_id from the
1443        // payload and reject mismatches.
1444        let expected = lex_vcs::Attestation::with_timestamp(
1445            att.stage_id.clone(),
1446            att.op_id.clone(),
1447            att.intent_id.clone(),
1448            att.kind.clone(),
1449            att.result.clone(),
1450            att.produced_by.clone(),
1451            att.cost.clone(),
1452            att.timestamp,
1453        ).attestation_id;
1454        if expected != att.attestation_id {
1455            return error_with_detail(409, "AttestationIdMismatch", serde_json::json!({
1456                "supplied": att.attestation_id,
1457                "expected": expected,
1458            }));
1459        }
1460        // The op_id field, if set, must point at an op the remote
1461        // knows about. Without this check, attestations would
1462        // dangle into a future sync that never lands the op.
1463        if let Some(op_id) = &att.op_id {
1464            match op_log.get(op_id) {
1465                Ok(Some(_)) => {}
1466                Ok(None) => return error_with_detail(422, "UnknownOp", serde_json::json!({
1467                    "attestation_id": att.attestation_id,
1468                    "op_id": op_id,
1469                })),
1470                Err(e) => return error_response(500, format!("op log read: {e}")),
1471            }
1472        }
1473    }
1474
1475    // Persist. `AttestationLog::put` is idempotent on
1476    // `attestation_id` and the by-stage index is rewritten as a
1477    // marker file, also idempotent.
1478    let mut added = 0usize;
1479    let mut added_ids: Vec<&lex_vcs::AttestationId> = Vec::new();
1480    for att in &attestations {
1481        let already_present = matches!(log.get(&att.attestation_id), Ok(Some(_)));
1482        match log.put(att) {
1483            Ok(()) => {
1484                if !already_present {
1485                    added += 1;
1486                    added_ids.push(&att.attestation_id);
1487                }
1488            }
1489            Err(e) => return error_response(500, format!("attestation log write: {e}")),
1490        }
1491    }
1492
1493    json_response(200, &serde_json::json!({
1494        "received": attestations.len(),
1495        "added": added,
1496        "skipped": attestations.len() - added,
1497        "added_ids": added_ids,
1498    }))
1499}
1500
1501/// `GET /v1/branches/<name>/head` (#242 follow-up). Probe endpoint
1502/// the `lex op push` client uses to discover the remote head before
1503/// computing a delta against `OpLog::ops_since`.
1504///
1505/// Response: `{ "branch": "main", "head_op": Option<OpId> }`.
1506/// Returns 200 even when the branch doesn't exist locally — the
1507/// answer in that case is `head_op: null`, which is the right
1508/// signal for "send everything you have."
1509/// `GET /v1/ops/since?after=<op_id>&branch=<name>&limit=<n>` (#260).
1510/// Server endpoint for `lex op pull`.
1511///
1512/// Returns a JSON array of `OperationRecord`s reachable from
1513/// `branch.head_op` but not from `<after>`, sorted **oldest-first**
1514/// so the client can apply them in topological order without
1515/// re-sorting. Empty array when:
1516///
1517/// * The branch doesn't exist on the remote.
1518/// * The branch's `head_op` is `None`.
1519/// * `after == branch.head_op` (caller is already at the remote's head).
1520/// * `after` is *ahead of* the remote's head (caller is past the
1521///   remote — the symmetric "remote behind" case from #260).
1522///
1523/// `branch` defaults to `main`. `limit` caps the response — useful
1524/// for chunked pulls of large gaps; clients re-issue with the next
1525/// `after` once the prefix has landed.
1526///
1527/// Failure modes:
1528///
1529/// * `400` if the query string is malformed.
1530/// * `200` with `[]` for any of the empty-result cases above. "Caller
1531///   is already up to date" is a normal answer, not an error.
1532pub(crate) fn ops_since_handler(state: &State, query: &str)
1533    -> Response<std::io::Cursor<Vec<u8>>>
1534{
1535    let mut after: Option<String> = None;
1536    let mut branch = String::from("main");
1537    let mut limit: Option<usize> = None;
1538    for kv in query.split('&') {
1539        let Some((k, v)) = kv.split_once('=') else { continue };
1540        match k {
1541            "after" => after = Some(v.to_string()),
1542            "branch" => branch = v.to_string(),
1543            "limit" => {
1544                limit = Some(match v.parse::<usize>() {
1545                    Ok(n) => n,
1546                    Err(_) => return error_response(400,
1547                        format!("limit must be a positive integer, got `{v}`")),
1548                });
1549            }
1550            _ => {}
1551        }
1552    }
1553
1554    let store = state.store.lock().unwrap();
1555    let log = match lex_vcs::OpLog::open(store.root()) {
1556        Ok(l) => l,
1557        Err(e) => return error_response(500, format!("opening op log: {e}")),
1558    };
1559    let head = match store.get_branch(&branch) {
1560        Ok(Some(b)) => b.head_op,
1561        Ok(None) => None,
1562        Err(e) => return error_response(500, format!("get_branch: {e}")),
1563    };
1564    let Some(head) = head else {
1565        return json_response(200, &serde_json::json!([]));
1566    };
1567
1568    let ops_since = match log.ops_since(&head, after.as_ref()) {
1569        Ok(o) => o,
1570        Err(e) => return error_response(500, format!("ops_since: {e}")),
1571    };
1572    // ops_since walks newest-first; reverse so the client receives
1573    // oldest-first and can apply them in topological order with
1574    // `OpLog::put` straight through.
1575    let mut ops = ops_since;
1576    ops.reverse();
1577    if let Some(n) = limit {
1578        ops.truncate(n);
1579    }
1580
1581    json_response(200, &serde_json::to_value(&ops).unwrap_or_default())
1582}
1583
1584/// `GET /v1/attestations/since?after-op=<op_id>&limit=<n>` (#260).
1585/// Mirror of `ops_since_handler` for the attestation log.
1586///
1587/// Returns attestations whose `op_id` field is reachable from
1588/// **any** branch's head — not just one — and not in `after_op`'s
1589/// ancestry. The cross-branch fan-out matches the push side:
1590/// attestations are stage-keyed, not branch-keyed, so a single
1591/// "since this op" filter is the right shape.
1592///
1593/// Attestations with `op_id: None` (e.g. `Override`,
1594/// `ProducerBlock`) are always included — the cutoff doesn't apply.
1595/// `--limit` caps the response.
1596pub(crate) fn attestations_since_handler(state: &State, query: &str)
1597    -> Response<std::io::Cursor<Vec<u8>>>
1598{
1599    let mut after_op: Option<String> = None;
1600    let mut limit: Option<usize> = None;
1601    for kv in query.split('&') {
1602        let Some((k, v)) = kv.split_once('=') else { continue };
1603        match k {
1604            "after-op" => after_op = Some(v.to_string()),
1605            "limit" => {
1606                limit = Some(match v.parse::<usize>() {
1607                    Ok(n) => n,
1608                    Err(_) => return error_response(400,
1609                        format!("limit must be a positive integer, got `{v}`")),
1610                });
1611            }
1612            _ => {}
1613        }
1614    }
1615
1616    let store = state.store.lock().unwrap();
1617    let log = match store.attestation_log() {
1618        Ok(l) => l,
1619        Err(e) => return error_response(500, format!("opening attestation log: {e}")),
1620    };
1621
1622    // Build the exclude set: every op_id reachable from `after_op`,
1623    // inclusive. Attestations whose op_id is in this set were
1624    // already known to the caller.
1625    let exclude: std::collections::BTreeSet<String> = match &after_op {
1626        None => std::collections::BTreeSet::new(),
1627        Some(cutoff) => {
1628            let op_log = match lex_vcs::OpLog::open(store.root()) {
1629                Ok(l) => l,
1630                Err(e) => return error_response(500, format!("opening op log: {e}")),
1631            };
1632            match op_log.walk_back(cutoff, None) {
1633                Ok(records) => records.into_iter().map(|r| r.op_id).collect(),
1634                Err(_) => {
1635                    // Cutoff op doesn't exist on this remote. Treat
1636                    // as "no exclude" — caller will get every
1637                    // attestation. They may dedup client-side.
1638                    std::collections::BTreeSet::new()
1639                }
1640            }
1641        }
1642    };
1643
1644    let all = match log.list_all() {
1645        Ok(v) => v,
1646        Err(e) => return error_response(500, format!("listing attestations: {e}")),
1647    };
1648    let mut filtered: Vec<lex_vcs::Attestation> = all
1649        .into_iter()
1650        .filter(|a| match &a.op_id {
1651            Some(op_id) => !exclude.contains(op_id),
1652            // No op_id = doesn't participate in the cutoff; always
1653            // ship it on the first pull, server-side idempotency
1654            // dedupes on the client.
1655            None => true,
1656        })
1657        .collect();
1658    // Stable order: oldest-first by `timestamp`, then by
1659    // `attestation_id` for ties. Lets the client land them
1660    // deterministically.
1661    filtered.sort_by(|a, b| {
1662        a.timestamp.cmp(&b.timestamp)
1663            .then_with(|| a.attestation_id.cmp(&b.attestation_id))
1664    });
1665    if let Some(n) = limit {
1666        filtered.truncate(n);
1667    }
1668
1669    json_response(200, &serde_json::to_value(&filtered).unwrap_or_default())
1670}
1671
1672// ── Package concept (#4) ────────────────────────────────────────────────────
1673
1674/// Full coordinates of one declared dependency, captured from the releaser's
1675/// `lex.toml` so the rendered archive can reproduce a faithful `[dependencies]`
1676/// table — carrying BOTH a vcs (`registry` + `version`) reference and a `git`
1677/// mirror when the source declared both. Every field is optional so a bare
1678/// registry dep, a bare git dep, or a dual dep all round-trip. `#[serde(default,
1679/// skip_serializing_if)]` keeps the record compact and back-compatible.
1680#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
1681struct DepSpec {
1682    #[serde(default, skip_serializing_if = "Option::is_none")]
1683    registry: Option<String>,
1684    #[serde(default, skip_serializing_if = "Option::is_none")]
1685    version: Option<String>,
1686    #[serde(default, skip_serializing_if = "Option::is_none")]
1687    git: Option<String>,
1688    #[serde(default, skip_serializing_if = "Option::is_none")]
1689    branch: Option<String>,
1690    #[serde(default, skip_serializing_if = "Option::is_none")]
1691    tag: Option<String>,
1692    #[serde(default, skip_serializing_if = "Option::is_none")]
1693    rev: Option<String>,
1694    #[serde(default, skip_serializing_if = "Option::is_none")]
1695    path: Option<String>,
1696}
1697
1698impl DepSpec {
1699    /// Render this spec as the body of a `lex.toml` inline dependency table,
1700    /// e.g. `{ registry = "…", version = "^0.9", git = "…" }`. Keys are emitted
1701    /// in a stable order (vcs primary first, then the git mirror) so the
1702    /// rendered manifest is deterministic.
1703    fn to_toml_inline(&self) -> Option<String> {
1704        let mut parts: Vec<String> = Vec::new();
1705        let mut push = |k: &str, v: &Option<String>| {
1706            if let Some(val) = v {
1707                parts.push(format!("{k} = {}", toml_str(val)));
1708            }
1709        };
1710        push("registry", &self.registry);
1711        push("version", &self.version);
1712        push("git", &self.git);
1713        push("branch", &self.branch);
1714        push("tag", &self.tag);
1715        push("rev", &self.rev);
1716        push("path", &self.path);
1717        if parts.is_empty() {
1718            None
1719        } else {
1720            Some(format!("{{ {} }}", parts.join(", ")))
1721        }
1722    }
1723}
1724
1725/// Quote a string as a TOML basic string (escaping `\` and `"`).
1726fn toml_str(s: &str) -> String {
1727    format!("\"{}\"", s.replace('\\', "\\\\").replace('"', "\\\""))
1728}
1729
1730/// Per-version record stored at `{store_root}/packages/{name}/{version}.json`.
1731#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1732struct PkgRecord {
1733    name: String,
1734    version: String,
1735    head_op: Option<String>,
1736    published_at: u64,
1737    /// Function names introduced or updated by this version (for retract).
1738    function_names: Vec<String>,
1739    /// External package dependencies of this release — declared by the
1740    /// releaser (from `lex.toml`) and/or extracted from the head's
1741    /// non-inlined imports. The edge set of the dependency graph (#893
1742    /// propagation). `#[serde(default)]` keeps pre-existing records readable.
1743    #[serde(default)]
1744    dependencies: Vec<String>,
1745    /// Full dependency coordinates (name → git+vcs refs) captured from the
1746    /// releaser's `lex.toml`, so the rendered install archive reproduces a
1747    /// faithful `[dependencies]` table (with both git and vcs references).
1748    /// `#[serde(default)]` keeps pre-existing records (which lack it) readable;
1749    /// when empty, the archive falls back to a bare `[package]` manifest.
1750    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
1751    dependency_specs: std::collections::BTreeMap<String, DepSpec>,
1752    /// Raw op JSON from each file in this publish.
1753    ops: Vec<serde_json::Value>,
1754}
1755
1756/// Whether a package is reachable without authentication.
1757///
1758/// Per-package (applies across all versions), GitHub-style: an org's
1759/// store can hold a mix of public and private packages. Defaults to
1760/// `Private`, so an index written before this field existed (or any
1761/// freshly published package) is private until an owner opts in.
1762#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, Default)]
1763#[serde(rename_all = "lowercase")]
1764pub enum Visibility {
1765    #[default]
1766    Private,
1767    Public,
1768}
1769
1770/// Index stored at `{store_root}/packages/{name}/index.json`.
1771///
1772/// Tracks which versions have been published and which is "latest", so
1773/// consumers can resolve `{name}@latest` without listing every file.
1774#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)]
1775struct PkgIndex {
1776    /// The most recently published version string (human label, not OpId).
1777    latest: Option<String>,
1778    /// All published versions, newest-last.
1779    versions: Vec<PkgVersionSummary>,
1780    /// Public/private flag. `#[serde(default)]` keeps pre-existing
1781    /// `index.json` files (which lack the field) deserializing as
1782    /// `Private`.
1783    #[serde(default)]
1784    visibility: Visibility,
1785}
1786
1787#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1788struct PkgVersionSummary {
1789    version: String,
1790    head_op: Option<String>,
1791    published_at: u64,
1792}
1793
1794fn pkg_name_dir(root: &std::path::Path, name: &str) -> PathBuf {
1795    root.join("packages").join(name)
1796}
1797
1798fn pkg_index_path(root: &std::path::Path, name: &str) -> PathBuf {
1799    pkg_name_dir(root, name).join("index.json")
1800}
1801
1802fn pkg_version_path(root: &std::path::Path, name: &str, version: &str) -> PathBuf {
1803    pkg_name_dir(root, name).join(format!("{version}.json"))
1804}
1805
1806fn pkg_archive_path(root: &std::path::Path, name: &str, version: &str) -> PathBuf {
1807    pkg_name_dir(root, name).join(format!("{version}.tar.gz"))
1808}
1809
1810fn load_pkg_index(root: &std::path::Path, name: &str) -> Option<PkgIndex> {
1811    let bytes = std::fs::read(pkg_index_path(root, name)).ok()?;
1812    serde_json::from_slice(&bytes).ok()
1813}
1814
1815fn load_pkg_record(root: &std::path::Path, name: &str, version: &str) -> Option<PkgRecord> {
1816    let bytes = std::fs::read(pkg_version_path(root, name, version)).ok()?;
1817    serde_json::from_slice(&bytes).ok()
1818}
1819
1820fn load_latest_pkg_record(root: &std::path::Path, name: &str) -> Option<PkgRecord> {
1821    let index = load_pkg_index(root, name)?;
1822    let latest = index.latest.clone()?;
1823    load_pkg_record(root, name, &latest)
1824}
1825
1826/// A package is public iff its index exists and is marked `Public`.
1827/// A missing index (unknown package) is treated as private, so the
1828/// public surface never distinguishes "private" from "does not exist".
1829fn pkg_is_public(root: &std::path::Path, name: &str) -> bool {
1830    load_pkg_index(root, name).map(|i| i.visibility) == Some(Visibility::Public)
1831}
1832
1833/// Reject package/version path segments that could escape the
1834/// `packages/` directory or otherwise aren't valid names. Mirrors the
1835/// tenant-id guard's spirit (defense in depth — lex-hub validates the
1836/// tenant, this validates the package/version).
1837fn valid_pkg_segment(s: &str) -> bool {
1838    !s.is_empty()
1839        && s.len() <= 128
1840        && s != "."
1841        && s != ".."
1842        && s.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
1843}
1844
1845#[derive(Deserialize)]
1846struct VisibilityReq {
1847    visibility: Visibility,
1848}
1849
1850/// `PUT /v1/pkg/{name}/visibility` — set a package public or private.
1851///
1852/// Authorization is implicit: this runs against a single tenant's store,
1853/// and the caller only reaches *this* store because the front door
1854/// (lex-hub) authenticated their token and selected it. A caller can
1855/// therefore only change visibility of packages they own.
1856fn pkg_set_visibility_handler(
1857    state: &State,
1858    name: &str,
1859    body: &str,
1860) -> Response<std::io::Cursor<Vec<u8>>> {
1861    if !valid_pkg_segment(name) {
1862        return error_response(400, format!("invalid package name {name:?}"));
1863    }
1864    let req: VisibilityReq = match serde_json::from_str(body) {
1865        Ok(r) => r,
1866        Err(e) => return error_response(400, format!("bad request: {e}")),
1867    };
1868    let mut index = match load_pkg_index(&state.root, name) {
1869        Some(i) => i,
1870        None => return error_response(404, format!("package {name:?} not found")),
1871    };
1872    index.visibility = req.visibility;
1873    let bytes = serde_json::to_vec_pretty(&index).unwrap_or_default();
1874    match std::fs::write(pkg_index_path(&state.root, name), bytes) {
1875        Ok(()) => json_response(
1876            200,
1877            &serde_json::json!({ "name": name, "visibility": index.visibility }),
1878        ),
1879        Err(e) => error_response(500, format!("write index: {e}")),
1880    }
1881}
1882
1883#[derive(serde::Deserialize)]
1884struct ReleaseReq {
1885    version: String,
1886    #[serde(default)]
1887    branch: Option<String>,
1888    /// External package dependencies (from the releaser's `lex.toml`).
1889    /// Unioned with any non-inlined external imports found in the head.
1890    #[serde(default)]
1891    dependencies: Vec<String>,
1892    /// Full dependency coordinates (name → git+vcs refs) from the releaser's
1893    /// `lex.toml`, so the rendered install archive reproduces a faithful
1894    /// `[dependencies]` table. Optional (back-compatible); when omitted, the
1895    /// archive manifest carries only `[package]`.
1896    #[serde(default)]
1897    dependency_specs: std::collections::BTreeMap<String, DepSpec>,
1898}
1899
1900/// `POST /v1/pkg/{name}/release` — cut an immutable versioned release of
1901/// an **op-log-hosted** package: snapshot the current branch head as
1902/// `name@version` in the registry (#893). Unlike `POST /v1/pkg/publish`
1903/// (archive upload), this records only the op-log ref (`head_op`) — a
1904/// consumer resolves the version, `op pull`s that head, and renders
1905/// source. A published version is immutable: re-releasing an existing
1906/// version is a 409, so a resolved+locked dependency can never change
1907/// under a consumer.
1908fn pkg_release_handler(state: &State, name: &str, body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
1909    if !valid_pkg_segment(name) {
1910        return error_response(400, format!("invalid package name {name:?}"));
1911    }
1912    let req: ReleaseReq = match serde_json::from_str(body) {
1913        Ok(r) => r,
1914        Err(e) => return error_response(400, format!("bad request: {e}")),
1915    };
1916    let version = req.version.trim().to_string();
1917    if version.is_empty() || !valid_pkg_segment(&version) {
1918        return error_response(400, "version must be a non-empty, path-safe string (e.g. 1.2.0)");
1919    }
1920    // Immutable: a published version can never be overwritten.
1921    if load_pkg_record(&state.root, name, &version).is_some() {
1922        return error_response(
1923            409,
1924            format!("{name}@{version} already released; releases are immutable — bump the version"),
1925        );
1926    }
1927
1928    let store = state.store.lock().unwrap();
1929    let branch = req.branch.unwrap_or_else(|| store.current_branch());
1930    let head_op = match store.get_branch(&branch) {
1931        Ok(Some(b)) => b.head_op,
1932        Ok(None) => return error_response(404, format!("unknown branch {branch:?}")),
1933        Err(e) => return error_response(500, format!("get_branch: {e}")),
1934    };
1935    let Some(head_op) = head_op else {
1936        return error_response(400, format!("branch {branch:?} has no commits to release"));
1937    };
1938
1939    // Version-bump gate (#893): the version increment must be at least what
1940    // the public-API change requires — a breaking change (a removed or
1941    // re-signatured public declaration) needs a *major* bump, an addition at
1942    // least a *minor*. This is what makes `^`/`~` resolution safe: a caret
1943    // update can't silently pull a breaking change mislabeled as a patch.
1944    //
1945    // Compare against the **semver predecessor** — the highest already-published
1946    // version strictly less than the new one (not merely `latest`, so releasing
1947    // 2.0.0 after 1.5.0 diffs against 1.5.0). No predecessor (first version, or
1948    // a back-port below everything) means nothing to gate.
1949    let predecessor = load_pkg_index(&state.root, name)
1950        .map(|i| i.versions)
1951        .unwrap_or_default()
1952        .into_iter()
1953        .filter_map(|v| lex_syntax::semver::parse_exact(&v.version).map(|p| (p, v)))
1954        .filter(|(p, _)| lex_syntax::semver::parse_exact(&version).map(|n| *p < n).unwrap_or(false))
1955        .max_by_key(|(p, _)| *p)
1956        .map(|(_, v)| v);
1957    if let Some(prev) = predecessor {
1958        if let (Some(prev_head), Some(declared)) = (
1959            prev.head_op.clone(),
1960            lex_syntax::semver::bump_between(&prev.version, &version),
1961        ) {
1962            if let (Ok(prev_api), Ok(new_api)) = (
1963                lex_store::api::public_api_at_op(&store, &prev_head),
1964                lex_store::api::public_api_at_op(&store, &head_op),
1965            ) {
1966                use lex_store::api::ApiChange;
1967                use lex_syntax::semver::Bump;
1968                let (required, why) = match lex_store::api::classify_api_change(&prev_api, &new_api) {
1969                    ApiChange::Breaking(d) => (Bump::Major, d),
1970                    ApiChange::Additive(d) => (Bump::Minor, d),
1971                    ApiChange::None => (Bump::Patch, String::new()),
1972                };
1973                if declared < required {
1974                    let need = match required {
1975                        Bump::Major => "major",
1976                        Bump::Minor => "minor",
1977                        Bump::Patch => "patch",
1978                    };
1979                    return error_response(
1980                        422,
1981                        format!(
1982                            "version bump too small: {} → {version} is a {declared:?} bump, \
1983                             but the API change ({why}) requires a {need} bump",
1984                            prev.version
1985                        ),
1986                    );
1987                }
1988            }
1989        }
1990    }
1991
1992    // The package's exported function names at this head (for retract /
1993    // the catalog), read through the SigId the head names each by.
1994    let head = store.branch_head(&branch).unwrap_or_default();
1995    let pairs: Vec<(String, String)> = head.iter().map(|(s, st)| (s.clone(), st.clone())).collect();
1996    let function_names: Vec<String> = store
1997        .get_asts_for_sigs_bulk(&pairs)
1998        .into_iter()
1999        .filter_map(|r| r.ok())
2000        .filter_map(|s| match s {
2001            lex_ast::Stage::FnDecl(fd) => Some(fd.name),
2002            _ => None,
2003        })
2004        .collect();
2005    // Dependency-graph edges (#893 propagation): the releaser's declared deps
2006    // (from lex.toml) unioned with any external imports still visible in the
2007    // head (most are inlined at publish, so the declared list is primary).
2008    let mut deps: std::collections::BTreeSet<String> = req.dependencies.into_iter().collect();
2009    if let Ok(extracted) = lex_store::api::external_dependencies_at_op(&store, &head_op) {
2010        deps.extend(extracted);
2011    }
2012    // Names captured as full coordinates count as dependency-graph edges too,
2013    // so a release that sends only `dependency_specs` still records the edges.
2014    deps.extend(req.dependency_specs.keys().cloned());
2015    let dependencies: Vec<String> = deps.into_iter().collect();
2016    drop(store);
2017
2018    let published_at = std::time::SystemTime::now()
2019        .duration_since(std::time::UNIX_EPOCH)
2020        .map(|d| d.as_secs())
2021        .unwrap_or(0);
2022    let record = PkgRecord {
2023        name: name.to_string(),
2024        version: version.clone(),
2025        head_op: Some(head_op.clone()),
2026        published_at,
2027        function_names,
2028        dependencies,
2029        dependency_specs: req.dependency_specs,
2030        ops: Vec::new(),
2031    };
2032    if let Err(e) = save_pkg_record(&state.root, &record, None) {
2033        return error_response(500, format!("write release: {e}"));
2034    }
2035    json_response(
2036        201,
2037        &serde_json::json!({
2038            "name": name,
2039            "version": version,
2040            "head_op": head_op,
2041            "branch": branch,
2042        }),
2043    )
2044}
2045
2046/// Names of the tenant's PUBLIC packages, sorted. Pure (filesystem in,
2047/// names out) so the visibility filter is unit-testable.
2048fn public_pkg_names(root: &std::path::Path) -> Vec<String> {
2049    list_pkg_names(root)
2050        .into_iter()
2051        .filter(|name| pkg_is_public(root, name))
2052        .collect()
2053}
2054
2055/// `GET /v1/public/<tenant>` (org page) — list only the tenant's PUBLIC
2056/// packages (latest version of each). Private packages are omitted, so
2057/// their existence is not revealed.
2058fn public_pkg_list_handler(state: &State) -> Response<std::io::Cursor<Vec<u8>>> {
2059    let packages: Vec<serde_json::Value> = public_pkg_names(&state.root)
2060        .iter()
2061        .filter_map(|name| {
2062            let r = load_latest_pkg_record(&state.root, name)?;
2063            Some(serde_json::json!({
2064                "name": r.name,
2065                "version": r.version,
2066                "head_op": r.head_op,
2067                "published_at": r.published_at,
2068            }))
2069        })
2070        .collect();
2071    json_response(200, &serde_json::json!({ "packages": packages }))
2072}
2073
2074/// A resolved public read target. Separated from response formatting so
2075/// the routing/guard logic is unit-testable without constructing HTTP
2076/// responses. `Err(status)` is a guard failure (405 = non-GET,
2077/// 404 = invalid/unknown route).
2078#[derive(Debug, PartialEq, Eq)]
2079enum PublicTarget {
2080    List,
2081    Latest(String),
2082    Versions(String),
2083    ApiDiff(String),
2084    Head(String),
2085    Version(String, String),
2086    Archive(String, String),
2087}
2088
2089impl PublicTarget {
2090    /// The package name a target refers to, if any (`List` has none).
2091    fn pkg_name(&self) -> Option<&str> {
2092        match self {
2093            PublicTarget::List => None,
2094            PublicTarget::Latest(n)
2095            | PublicTarget::Versions(n)
2096            | PublicTarget::ApiDiff(n)
2097            | PublicTarget::Head(n)
2098            | PublicTarget::Version(n, _)
2099            | PublicTarget::Archive(n, _) => Some(n),
2100        }
2101    }
2102}
2103
2104/// Pure routing decision for `/v1/public/<tenant>` reads. `path` is the
2105/// portion after the tenant (leading `/` ok). Enforces GET-only and
2106/// per-segment name validation; does NOT consult the store (visibility
2107/// is checked by the caller, which has store access).
2108fn resolve_public(method: &Method, path: &str) -> Result<PublicTarget, u16> {
2109    if !matches!(method, Method::Get) {
2110        return Err(405);
2111    }
2112    let rest = path.trim_matches('/');
2113    if rest.is_empty() {
2114        return Ok(PublicTarget::List);
2115    }
2116    let segs: Vec<&str> = rest.split('/').collect();
2117    if !segs.iter().all(|s| valid_pkg_segment(s)) {
2118        return Err(404);
2119    }
2120    match segs.as_slice() {
2121        [n] => Ok(PublicTarget::Latest(n.to_string())),
2122        [n, "versions"] => Ok(PublicTarget::Versions(n.to_string())),
2123        [n, "api-diff"] => Ok(PublicTarget::ApiDiff(n.to_string())),
2124        [n, "head"] => Ok(PublicTarget::Head(n.to_string())),
2125        [n, v, "archive"] => Ok(PublicTarget::Archive(n.to_string(), v.to_string())),
2126        [n, v] => Ok(PublicTarget::Version(n.to_string(), v.to_string())),
2127        _ => Err(404),
2128    }
2129}
2130
2131/// Unauthenticated, read-only access to **public** packages in `state`'s
2132/// store. `path` is the portion of the URL after `/v1/public/<tenant>`
2133/// (with a leading `/`); lex-hub resolves `<tenant>` → store and calls
2134/// this. Visibility gating, GET-only enforcement, and segment validation
2135/// all live here so the whole public surface is auditable in one place.
2136///
2137/// Everything served here is package-scoped — manifests and the source
2138/// archive of a public package's own publish — so it cannot leak code
2139/// from a private package that happens to share content-addressed stages
2140/// in the same store.
2141pub fn route_public(
2142    state: &State,
2143    method: &Method,
2144    path: &str,
2145    query: &str,
2146) -> Response<std::io::Cursor<Vec<u8>>> {
2147    let target = match resolve_public(method, path) {
2148        Ok(t) => t,
2149        Err(405) => return error_response(405, "public read is GET-only"),
2150        Err(_) => return error_response(404, "not found"),
2151    };
2152    // The org listing already filters to public packages itself.
2153    if let PublicTarget::List = target {
2154        return public_pkg_list_handler(state);
2155    }
2156    // Single 404 for both "private" and "absent" — never reveal which.
2157    if let Some(name) = target.pkg_name() {
2158        if !pkg_is_public(&state.root, name) {
2159            return error_response(404, format!("package {name:?} not found"));
2160        }
2161    }
2162    match target {
2163        PublicTarget::List => unreachable!("handled above"),
2164        PublicTarget::Latest(n) => pkg_get_handler(state, &n),
2165        PublicTarget::Versions(n) => pkg_versions_handler(state, &n),
2166        PublicTarget::ApiDiff(n) => pkg_api_diff_handler(state, &n, query),
2167        PublicTarget::Head(n) => pkg_head_handler(state, &n),
2168        PublicTarget::Version(n, v) => pkg_get_version_handler(state, &n, &v),
2169        PublicTarget::Archive(n, v) => pkg_archive_handler(state, &n, &v),
2170    }
2171}
2172
2173fn save_pkg_record(
2174    root: &std::path::Path,
2175    record: &PkgRecord,
2176    // `None` for an op-log release, whose source of truth is the op-log at
2177    // `head_op` (pull + render); `Some` for an archive-upload publish.
2178    archive: Option<&[u8]>,
2179) -> std::io::Result<()> {
2180    let dir = pkg_name_dir(root, &record.name);
2181    std::fs::create_dir_all(&dir)?;
2182
2183    // Per-version record.
2184    let rec_bytes = serde_json::to_vec_pretty(record).unwrap_or_default();
2185    std::fs::write(pkg_version_path(root, &record.name, &record.version), rec_bytes)?;
2186
2187    // Archive (tar.gz) for the download endpoint, when one was uploaded.
2188    if let Some(archive) = archive {
2189        std::fs::write(pkg_archive_path(root, &record.name, &record.version), archive)?;
2190    }
2191
2192    // Update the index.
2193    let mut index = load_pkg_index(root, &record.name).unwrap_or_default();
2194    index.latest = Some(record.version.clone());
2195    if !index.versions.iter().any(|v| v.version == record.version) {
2196        index.versions.push(PkgVersionSummary {
2197            version: record.version.clone(),
2198            head_op: record.head_op.clone(),
2199            published_at: record.published_at,
2200        });
2201    }
2202    let idx_bytes = serde_json::to_vec_pretty(&index).unwrap_or_default();
2203    std::fs::write(pkg_index_path(root, &record.name), idx_bytes)
2204}
2205
2206fn list_pkg_names(root: &std::path::Path) -> Vec<String> {
2207    let dir = root.join("packages");
2208    let Ok(entries) = std::fs::read_dir(&dir) else {
2209        return Vec::new();
2210    };
2211    let mut names: Vec<String> = entries
2212        .filter_map(|e| e.ok())
2213        .filter(|e| e.path().is_dir())
2214        .filter_map(|e| e.file_name().into_string().ok())
2215        .collect();
2216    names.sort();
2217    names
2218}
2219
2220fn collect_lex_files(dir: &std::path::Path, out: &mut Vec<PathBuf>) {
2221    let Ok(entries) = std::fs::read_dir(dir) else { return };
2222    let mut entries: Vec<_> = entries.filter_map(|e| e.ok()).collect();
2223    entries.sort_by_key(|e| e.path());
2224    for entry in entries {
2225        let path = entry.path();
2226        if path.is_dir() {
2227            collect_lex_files(&path, out);
2228        } else if path.extension().and_then(|x| x.to_str()) == Some("lex") {
2229            out.push(path);
2230        }
2231    }
2232}
2233
2234/// `POST /v1/pkg/publish` — publish a multi-file package from a `.tar.gz`
2235/// archive containing `lex.toml` and `src/**/*.lex`.
2236fn pkg_publish_handler(state: &State, body: &[u8]) -> Response<std::io::Cursor<Vec<u8>>> {
2237    let tmp = match tempfile::TempDir::new() {
2238        Ok(t) => t,
2239        Err(e) => return error_response(500, format!("create temp dir: {e}")),
2240    };
2241    {
2242        let gz = flate2::read::GzDecoder::new(std::io::Cursor::new(body));
2243        let mut ar = tar::Archive::new(gz);
2244        if let Err(e) = ar.unpack(tmp.path()) {
2245            return error_response(400, format!("unpack archive: {e}"));
2246        }
2247    }
2248
2249    let toml_path = tmp.path().join("lex.toml");
2250    if !toml_path.exists() {
2251        return error_response(400, "archive must contain lex.toml at root");
2252    }
2253    let manifest = match Manifest::load(&toml_path) {
2254        Ok(m) => m,
2255        Err(e) => return error_response(400, format!("lex.toml: {e}")),
2256    };
2257    let (pkg_name, pkg_version) = match &manifest.package {
2258        Some(m) => (m.name.clone(), m.version.clone()),
2259        None => return error_response(400, "lex.toml must have a [package] section"),
2260    };
2261
2262    // Reject a same-(name, version) re-publish BEFORE any store write:
2263    // this ran after the publish loop, so a 409'd duplicate still
2264    // appended its ops to the tenant's op log (#826).
2265    if load_pkg_record(&state.root, &pkg_name, &pkg_version).is_some() {
2266        return error_response(
2267            409,
2268            format!(
2269                "package {pkg_name}@{pkg_version} already published; \
2270                 bump the version in lex.toml to publish a new release"
2271            ),
2272        );
2273    }
2274
2275    let src_dir = tmp.path().join("src");
2276    if !src_dir.exists() {
2277        return error_response(400, "archive must contain a src/ directory");
2278    }
2279    let mut lex_files: Vec<PathBuf> = Vec::new();
2280    collect_lex_files(&src_dir, &mut lex_files);
2281    if lex_files.is_empty() {
2282        return error_response(400, "no .lex files found in src/");
2283    }
2284
2285    let store = state.store.lock().unwrap();
2286    let branch = store.current_branch();
2287
2288    // `old_fns_by_name` mirrors the branch's current function set,
2289    // GROUPED by name rather than collapsed to one entry per name. The
2290    // branch is tenant-wide and carries history: several live functions
2291    // can share a bare name (#818 — three unrelated `validate`s across
2292    // `field.lex`/`schema.lex`/`validator.lex` in a real package, all
2293    // published before names were file-prefixed). `SigId` disambiguates
2294    // them correctly — it hashes the full signature, not just the name —
2295    // so the bug was ever collapsing multiple SigIds sharing a name down
2296    // to one `FnDecl`, silently discarding the others and corrupting
2297    // later diffs against them. The package side can no longer produce
2298    // such a collision: one pass, one prefix-mangled name per
2299    // declaration (#828).
2300    //
2301    // Reading it once, outside any per-file loop, is what fixed #813:
2302    // `branch_head` walks the whole branch op history and `get_ast` is a
2303    // disk fetch per live function, so re-deriving this per file was
2304    // O(files * (history_size + live_fn_count)) — tens of minutes on a
2305    // tenant with 110k+ accumulated ops. Since #828 there is one pass, so
2306    // the shape is structural rather than a discipline to maintain.
2307    let old_head = match store.branch_head(&branch) {
2308        Ok(h) => h,
2309        Err(e) => return error_response(500, format!("branch_head: {e}")),
2310    };
2311    // Resolve each AST through the SigId the branch head names, never its
2312    // StageId: StageIds are name-independent, so two live functions
2313    // differing only in name share one, `stage_index` maps it to just one
2314    // of their sigs, and the name that lookup missed got re-reported as an
2315    // Add on every publish of unchanged source (#826). It also skips that
2316    // index (#825) — see `get_asts_for_sigs_bulk`.
2317    let old_pairs: Vec<(String, String)> =
2318        old_head.iter().map(|(sig, stage)| (sig.clone(), stage.clone())).collect();
2319    let mut old_fns_by_name: BTreeMap<String, Vec<lex_ast::FnDecl>> = BTreeMap::new();
2320    for fd in store.get_asts_for_sigs_bulk(&old_pairs)
2321        .into_iter()
2322        .filter_map(|r| r.ok())
2323        .filter_map(|s| match s { lex_ast::Stage::FnDecl(fd) => Some(fd), _ => None })
2324    {
2325        old_fns_by_name.entry(fd.name.clone()).or_default().push(fd);
2326    }
2327    // Old `type`s on the branch, by (mangled) name — captured too (#895).
2328    // Types aren't overloaded, so a plain name map needs no take_matching.
2329    let mut old_types_by_name: BTreeMap<String, lex_ast::TypeDecl> = BTreeMap::new();
2330    for td in store.get_asts_for_sigs_bulk(&old_pairs)
2331        .into_iter()
2332        .filter_map(|r| r.ok())
2333        .filter_map(|s| match s { lex_ast::Stage::TypeDecl(td) => Some(td), _ => None })
2334    {
2335        old_types_by_name.insert(td.name.clone(), td);
2336    }
2337    // A name-independent fingerprint of a function's *contract*
2338    // (effects, param types, return type, examples — everything SigId
2339    // hashes except the name). Used only to disambiguate when multiple
2340    // candidates share a bare name: if this file's own declaration
2341    // structurally matches exactly one of them, that's unambiguously
2342    // the same evolving function; otherwise it's a new, unrelated
2343    // declaration that happens to reuse a name used elsewhere.
2344    fn structural_key(fd: &lex_ast::FnDecl) -> Option<String> {
2345        let mut anon = fd.clone();
2346        anon.name = String::new();
2347        lex_ast::sig_id(&lex_ast::Stage::FnDecl(anon))
2348    }
2349
2350    // Look up (and consume) the candidate in `map[name]` that matches
2351    // `new_fd`'s identity. One candidate is taken as unambiguous (so a
2352    // signature-changing edit still reads as a modification of the same
2353    // function); several always require an exact structural match to
2354    // disambiguate, and no match means this is a distinct, unrelated
2355    // declaration reusing a name used elsewhere — `None` rather than a
2356    // guess that mis-attributes history. Safe because the candidate pool
2357    // is complete and never grows: it is built once, up front, from the
2358    // whole branch.
2359    fn take_matching(
2360        map: &mut BTreeMap<String, Vec<lex_ast::FnDecl>>,
2361        name: &str,
2362        new_fd: &lex_ast::FnDecl,
2363    ) -> Option<lex_ast::FnDecl> {
2364        let candidates = map.get_mut(name)?;
2365        let idx = match candidates.len() {
2366            0 => return None,
2367            1 => 0,
2368            _ => {
2369                let want = structural_key(new_fd);
2370                candidates.iter().position(|c| structural_key(c) == want)?
2371            }
2372        };
2373        let matched = candidates.remove(idx);
2374        if candidates.is_empty() {
2375            map.remove(name);
2376        }
2377        Some(matched)
2378    }
2379
2380    // ---- one pass over the whole package -------------------------
2381    // `load_package` merges every file in the archive into ONE program
2382    // through a single shared loader pass. The flattening per-file loads
2383    // this replaced gave each top-level file its own copy of everything
2384    // it imported, so a shared dependency was canonicalized,
2385    // type-checked, diffed and published once per importing file: the
2386    // real 21-file `lex-schema` package, whose `error.lex` is imported by
2387    // 17 of its files, produced 2,239 `FnDecl`s for 693 distinct names
2388    // and paid for all 2,239 (#828). Collapsing 21 `publish_program`
2389    // calls into one matters even more than the 3.2x itself, because each
2390    // call independently reads every live function on the branch and
2391    // walks the op log for `old_imports`.
2392    //
2393    // The cost is that nothing is published under its bare source name
2394    // any more: `fn validate` in `src/field.lex` is
2395    // `field_<hash>.validate`. That is what makes one program safe to
2396    // check as a unit — the checker's global scope is keyed by name, and
2397    // two files may each declare their own `validate` (#818) — and it is
2398    // the naming change #828 asks for in exchange for the single pass.
2399    // The archive-publish path type-checks the loaded program with no
2400    // dependency resolver, so it still inlines registry/git deps to stay
2401    // self-contained (#930). The op-log-native `op push` path publishes
2402    // without inlining and resolves at the gate instead.
2403    let loaded = match load_package(&lex_files, tmp.path(), &pkg_name, /*inline_packages=*/ true) {
2404        Ok(p) => p,
2405        Err(e) => return error_response(400, format!("load package: {e}")),
2406    };
2407    let mut stages = canonicalize_program(&loaded.program);
2408    // Type errors are reported for the package, not per file: every name
2409    // in them carries its file's mangling prefix, so the offending file
2410    // is still named. Nothing is published unless the whole package
2411    // checks, where before each file published as it was processed and a
2412    // later failure left the earlier files' ops applied.
2413    if let Err(errs) = lex_types::check_and_rewrite_program(&mut stages) {
2414        return error_with_detail(
2415            422,
2416            format!("type errors in package {pkg_name}"),
2417            serde_json::to_value(&errs).unwrap(),
2418        );
2419    }
2420    let new_fns = stage_fns(&stages);
2421    let all_function_names: Vec<String> = new_fns.keys().cloned().collect();
2422
2423    // Resolve each declaration against the branch's current state. No
2424    // in-request bookkeeping is needed now: one pass means each name is
2425    // declared once, so there is no earlier file in this request whose
2426    // just-published version a later one has to diff against.
2427    let mut old_fns: BTreeMap<String, lex_ast::FnDecl> = BTreeMap::new();
2428    for (name, new_fd) in &new_fns {
2429        if let Some(fd) = take_matching(&mut old_fns_by_name, name, new_fd) {
2430            old_fns.insert(name.clone(), fd);
2431        }
2432    }
2433    let new_types = stage_types(&stages);
2434    // Only diff old types this publish also declares — one on the branch
2435    // but absent from this file-set is left alone (as unclaimed old fns
2436    // are), not read as removed.
2437    let old_types: BTreeMap<String, lex_ast::TypeDecl> = new_types
2438        .keys()
2439        .filter_map(|n| old_types_by_name.get(n).map(|td| (n.clone(), td.clone())))
2440        .collect();
2441    let report =
2442        lex_vcs::compute_diff_with_types(&old_fns, &new_fns, &old_types, &new_types, false);
2443
2444    // Imports stay attributed per file — `AddImport`/`RemoveImport` carry
2445    // an `in_file`, and history records these same root-relative keys.
2446    // Each file now gets only the modules it imports itself; a flattened
2447    // per-file load could not tell those from its children's.
2448    //
2449    // `imports_by_file` carries each import's real `as` alias (#909/#930), so
2450    // a non-default `import "..." as x` round-trips as `x`.
2451    let mut new_imports = lex_vcs::ImportMap::new();
2452    for (file, modules) in &loaded.imports_by_file {
2453        let entry = new_imports.entry(file.clone()).or_default();
2454        for (reference, alias) in modules {
2455            entry.insert(lex_vcs::ImportRef {
2456                reference: reference.clone(),
2457                alias: alias.clone(),
2458            });
2459        }
2460    }
2461
2462    // Record each declaration's source file (#894) so an HTTP-published
2463    // package de-flattens in `export-git` too, not just a CLI-published one.
2464    let outcome = match store.publish_program_with_intent(
2465        &branch,
2466        &stages,
2467        &report,
2468        &new_imports,
2469        false,
2470        None,
2471        None,
2472        &loaded.module_prefixes,
2473    ) {
2474        Ok(outcome) => outcome,
2475        Err(lex_store::StoreError::TypeError(errs)) => {
2476            return error_with_detail(422, "type errors", serde_json::to_value(&errs).unwrap());
2477        }
2478        Err(e) => return write_error_response("publish_program", e),
2479    };
2480    let all_ops: Vec<serde_json::Value> = match serde_json::to_value(&outcome.ops) {
2481        Ok(serde_json::Value::Array(arr)) => arr,
2482        _ => Vec::new(),
2483    };
2484    let final_head_op = outcome.head_op;
2485
2486    // Deliberately no "genuinely removed" cleanup pass here. Whatever
2487    // remains in `old_fns_by_name` was never claimed by any file in
2488    // THIS archive — but the branch this walks is scoped to the whole
2489    // TENANT, not to this one package: a tenant that has ever published
2490    // more than one package (confirmed in production — `lex-schema` and
2491    // `lex-ocpi` share a tenant) has every other package's functions
2492    // sitting in `old_fns_by_name` too, forever unclaimed by any file in
2493    // *this* package's own archive. An earlier version of this handler
2494    // treated all such leftovers as "removed" and would have emitted
2495    // RemoveFunction ops for a completely unrelated package's functions
2496    // on every single publish. Caught before it shipped (`diff_to_ops`
2497    // failed atomically on a stale SigId before applying anything, so
2498    // no data was actually lost) — see alpibrusl/lex-lang#818's
2499    // follow-up. Nothing here currently tracks which package "owns" a
2500    // given branch function, so there's no reliable way to tell a
2501    // genuine same-package removal from another package's untouched
2502    // function; leaving a deleted function's stage un-removed (it just
2503    // sits there, unreferenced) is the safe default until package-scoped
2504    // ownership is tracked, not silently deleting a stranger's data.
2505    // Since #828 the same applies to a package's own previous names: the
2506    // first publish after file-prefixed naming landed leaves the bare
2507    // names it used to publish under sitting unreferenced, for the same
2508    // reason — this view cannot tell them from another package's.
2509
2510    let now = SystemTime::now()
2511        .duration_since(UNIX_EPOCH)
2512        .map(|d| d.as_secs())
2513        .unwrap_or(0);
2514    // External deps from the head's non-inlined imports (archive publish; the
2515    // op-log release route also accepts declared deps from lex.toml).
2516    let dependencies: Vec<String> = final_head_op
2517        .as_ref()
2518        .and_then(|h| lex_store::api::external_dependencies_at_op(&store, h).ok())
2519        .unwrap_or_default();
2520    let record = PkgRecord {
2521        name: pkg_name.clone(),
2522        version: pkg_version,
2523        head_op: final_head_op.clone(),
2524        published_at: now,
2525        function_names: all_function_names,
2526        dependencies,
2527        // The archive-upload path stores the uploaded archive verbatim (its
2528        // lex.toml already carries the full dependency table), so it is served
2529        // directly rather than re-rendered — no captured specs needed here.
2530        dependency_specs: Default::default(),
2531        ops: all_ops.clone(),
2532    };
2533    if let Err(e) = save_pkg_record(&state.root, &record, Some(body)) {
2534        return error_response(500, format!("save package index: {e}"));
2535    }
2536
2537    json_response(200, &serde_json::json!({
2538        "package": pkg_name,
2539        "ops": all_ops,
2540        "head_op": final_head_op,
2541    }))
2542}
2543
2544/// `GET /v1/pkg` — list packages published by this tenant (latest version of each).
2545fn pkg_list_handler(state: &State) -> Response<std::io::Cursor<Vec<u8>>> {
2546    let names = list_pkg_names(&state.root);
2547    let packages: Vec<serde_json::Value> = names.iter()
2548        .filter_map(|name| {
2549            let idx = load_pkg_index(&state.root, name)?;
2550            let latest = idx.latest.as_deref()?;
2551            let r = load_pkg_record(&state.root, name, latest)?;
2552            Some(serde_json::json!({
2553                "name": r.name,
2554                "version": r.version,
2555                "head_op": r.head_op,
2556                "published_at": r.published_at,
2557            }))
2558        })
2559        .collect();
2560    json_response(200, &serde_json::json!({ "packages": packages }))
2561}
2562
2563/// `GET /v1/pkg/{name}` — latest version details for a package.
2564fn pkg_get_handler(state: &State, name: &str) -> Response<std::io::Cursor<Vec<u8>>> {
2565    match load_latest_pkg_record(&state.root, name) {
2566        Some(r) => json_response(200, &serde_json::json!({
2567            "name": r.name,
2568            "version": r.version,
2569            "head_op": r.head_op,
2570            "published_at": r.published_at,
2571            "function_names": r.function_names,
2572            "ops": r.ops,
2573        })),
2574        None => error_response(404, format!("package {name:?} not found")),
2575    }
2576}
2577
2578/// `GET /v1/pkg/{name}/versions` — all published versions for a package.
2579fn pkg_versions_handler(state: &State, name: &str) -> Response<std::io::Cursor<Vec<u8>>> {
2580    match load_pkg_index(&state.root, name) {
2581        Some(idx) => json_response(200, &serde_json::json!({
2582            "name": name,
2583            "latest": idx.latest,
2584            "versions": idx.versions,
2585        })),
2586        None => error_response(404, format!("package {name:?} not found")),
2587    }
2588}
2589
2590/// `GET /v1/pkg/{name}/api-diff?from=<v>&to=<v>` — how the public API changed
2591/// between two releases (#893 propagation): the classification and the
2592/// mechanically-propagatable renames, so `lex propagate` can auto-derive its
2593/// `--rename` edits from a hosted release pair rather than have them restated.
2594fn pkg_api_diff_handler(state: &State, name: &str, query: &str) -> Response<std::io::Cursor<Vec<u8>>> {
2595    let mut from: Option<String> = None;
2596    let mut to: Option<String> = None;
2597    for kv in query.split('&') {
2598        match kv.split_once('=') {
2599            Some(("from", v)) => from = Some(v.to_string()),
2600            Some(("to", v)) => to = Some(v.to_string()),
2601            _ => {}
2602        }
2603    }
2604    let (Some(from), Some(to)) = (from, to) else {
2605        return error_response(400, "api-diff requires ?from=<version>&to=<version>");
2606    };
2607    let head_of = |v: &str| load_pkg_record(&state.root, name, v).and_then(|r| r.head_op);
2608    let (Some(from_head), Some(to_head)) = (head_of(&from), head_of(&to)) else {
2609        return error_response(404, format!("{name}: unknown release in {from}..{to}"));
2610    };
2611
2612    let store = state.store.lock().unwrap();
2613    let (prev_api, new_api) = match (
2614        lex_store::api::public_api_at_op(&store, &from_head),
2615        lex_store::api::public_api_at_op(&store, &to_head),
2616    ) {
2617        (Ok(a), Ok(b)) => (a, b),
2618        _ => return error_response(500, "could not read package APIs for the given releases"),
2619    };
2620    let (change, detail) = match lex_store::api::classify_api_change(&prev_api, &new_api) {
2621        lex_store::api::ApiChange::Breaking(d) => ("breaking", d),
2622        lex_store::api::ApiChange::Additive(d) => ("additive", d),
2623        lex_store::api::ApiChange::None => ("none", String::new()),
2624    };
2625    let renames = lex_store::api::detect_renames(&prev_api, &new_api);
2626    json_response(200, &serde_json::json!({
2627        "name": name, "from": from, "to": to,
2628        "change": change, "detail": detail, "renames": renames,
2629    }))
2630}
2631
2632/// `GET /v1/pkg/{name}/{version}` — specific version details.
2633fn pkg_get_version_handler(state: &State, name: &str, version: &str) -> Response<std::io::Cursor<Vec<u8>>> {
2634    match load_pkg_record(&state.root, name, version) {
2635        Some(r) => json_response(200, &serde_json::json!({
2636            "name": r.name,
2637            "version": r.version,
2638            "head_op": r.head_op,
2639            "published_at": r.published_at,
2640            "function_names": r.function_names,
2641            "dependencies": r.dependencies,
2642            "ops": r.ops,
2643        })),
2644        None => error_response(404, format!("package {name:?}@{version:?} not found")),
2645    }
2646}
2647
2648/// `GET /v1/pkg/{name}/{version}/archive` — download the source tar.gz.
2649fn pkg_archive_handler(state: &State, name: &str, version: &str) -> Response<std::io::Cursor<Vec<u8>>> {
2650    let gzip = |bytes: Vec<u8>| {
2651        Response::from_data(bytes).with_status_code(200).with_header(
2652            tiny_http::Header::from_bytes(&b"Content-Type"[..], &b"application/gzip"[..]).unwrap(),
2653        )
2654    };
2655
2656    // 1. A stored archive (the `lex pkg publish --registry` upload path).
2657    if let Ok(bytes) = std::fs::read(pkg_archive_path(&state.root, name, version)) {
2658        return gzip(bytes);
2659    }
2660
2661    // 2. An op-log-native release (op push + `POST …/release`, #911) has no
2662    //    stored archive — render one from the pinned op-log head so the
2663    //    package installs like any other (#920).
2664    if let Some(record) = load_pkg_record(&state.root, name, version) {
2665        if let Some(head_op) = record.head_op.clone() {
2666            match render_op_log_archive(state, name, version, &head_op, &record.dependency_specs) {
2667                Ok(bytes) => return gzip(bytes),
2668                Err(e) => {
2669                    return error_response(500, format!("rendering archive for {name:?}@{version:?}: {e}"));
2670                }
2671            }
2672        }
2673    }
2674
2675    error_response(404, format!("archive for {name:?}@{version:?} not found"))
2676}
2677
2678/// The archive path for a single-module head that records no source file of
2679/// its own — every op predating `in_file`. Historical, and kept so already
2680/// released packages in that state do not change name underneath their
2681/// dependents; the cure for them is a re-push, not a rename here (#988).
2682const DEFAULT_ARCHIVE_MODULE: &str = "src/lib.lex";
2683
2684/// Build a gzip-tar package archive (`lex.toml` + the package's `src/*.lex`)
2685/// by rendering the op-log head `head_op` to source — the composition of a
2686/// registry release (#911) with the op-log, so an `op push`-hosted package is
2687/// installable without a separately-uploaded archive (#920).
2688fn render_op_log_archive(
2689    state: &State,
2690    name: &str,
2691    version: &str,
2692    head_op: &str,
2693    dependency_specs: &std::collections::BTreeMap<String, DepSpec>,
2694) -> Result<Vec<u8>, String> {
2695    // De-flatten the head into its source tree — one file for a single-module
2696    // package, or the full `src/*.lex` layout for a multi-module one (#894).
2697    // The same renderer `lex export-git` uses, so the installed source matches
2698    // the git mirror. Either way the *names* come from the head itself; this
2699    // used to invent `src/lib.lex` for the single-module arm (#988).
2700    let files: Vec<(String, String)> = {
2701        let store = state.store.lock().unwrap();
2702        let head = lex_store::render::package_head_at_op(&store, head_op)
2703            .map_err(|e| format!("reading head {head_op}: {e}"))?;
2704        match lex_store::render::render_source(&store, &head)
2705            .map_err(|e| format!("rendering source at {head_op}: {e}"))?
2706        {
2707            lex_store::render::RenderedSource::Single { path, src } => {
2708                // A head that records no file keeps the registry's historical
2709                // name; one that does keeps its own (#988).
2710                vec![(path.unwrap_or_else(|| DEFAULT_ARCHIVE_MODULE.to_string()), src)]
2711            }
2712            lex_store::render::RenderedSource::Multi(tree) => tree.into_iter().collect(),
2713        }
2714    };
2715
2716    // Reconstruct the manifest from the release record. When the release
2717    // captured full dependency coordinates, emit a faithful `[dependencies]`
2718    // table (carrying both git and vcs references where the source declared
2719    // both); otherwise a bare `[package]` (pre-dependency-capture releases).
2720    let mut manifest = format!("[package]\nname = \"{name}\"\nversion = \"{version}\"\n");
2721    let dep_lines: Vec<String> = dependency_specs
2722        .iter()
2723        .filter_map(|(dep_name, spec)| spec.to_toml_inline().map(|inline| format!("{dep_name} = {inline}")))
2724        .collect();
2725    if !dep_lines.is_empty() {
2726        manifest.push_str("\n[dependencies]\n");
2727        for line in dep_lines {
2728            manifest.push_str(&line);
2729            manifest.push('\n');
2730        }
2731    }
2732    let mut enc = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
2733    {
2734        let mut ar = tar::Builder::new(&mut enc);
2735        let mut append = |p: &str, data: &[u8]| -> std::io::Result<()> {
2736            let mut h = tar::Header::new_gnu();
2737            h.set_size(data.len() as u64);
2738            h.set_mode(0o644);
2739            h.set_cksum();
2740            ar.append_data(&mut h, p, data)
2741        };
2742        append("lex.toml", manifest.as_bytes()).map_err(|e| e.to_string())?;
2743        for (path, src) in &files {
2744            append(path, src.as_bytes()).map_err(|e| e.to_string())?;
2745        }
2746        ar.finish().map_err(|e| e.to_string())?;
2747    }
2748    enc.finish().map_err(|e| e.to_string())
2749}
2750
2751/// `GET /v1/pkg/{name}/head` — head op for a package's latest version.
2752fn pkg_head_handler(state: &State, name: &str) -> Response<std::io::Cursor<Vec<u8>>> {
2753    match load_latest_pkg_record(&state.root, name) {
2754        Some(r) => json_response(200, &serde_json::json!({
2755            "name": r.name,
2756            "version": r.version,
2757            "head_op": r.head_op,
2758        })),
2759        None => error_response(404, format!("package {name:?} not found")),
2760    }
2761}
2762
2763/// `DELETE /v1/pkg/{name}` — retract the latest version of a package.
2764fn pkg_delete_handler(state: &State, name: &str) -> Response<std::io::Cursor<Vec<u8>>> {
2765    let record = match load_latest_pkg_record(&state.root, name) {
2766        Some(r) => r,
2767        None => return error_response(404, format!("package {name:?} not found")),
2768    };
2769
2770    let store = state.store.lock().unwrap();
2771    let branch = store.current_branch();
2772
2773    let head = match store.branch_head(&branch) {
2774        Ok(h) => h,
2775        Err(e) => return error_response(500, format!("branch_head: {e}")),
2776    };
2777
2778    // Build old_fns from this package's function names that are still on
2779    // the branch, reading each AST through the SigId the head names. A
2780    // StageId-keyed read is ambiguous when two live functions differ only
2781    // in name (#826), and here that ambiguity decides what gets REMOVED:
2782    // it could both miss one of this package's functions and match a name
2783    // belonging to another package sharing the stage.
2784    let head_pairs: Vec<(String, String)> = head
2785        .iter()
2786        .map(|(sig, stage)| (sig.clone(), stage.clone()))
2787        .collect();
2788    let old_fns: BTreeMap<String, lex_ast::FnDecl> = store
2789        .get_asts_for_sigs_bulk(&head_pairs)
2790        .into_iter()
2791        .filter_map(|r| r.ok())
2792        .filter_map(|s| match s {
2793            lex_ast::Stage::FnDecl(fd)
2794                if record.function_names.contains(&fd.name) => Some((fd.name.clone(), fd)),
2795            _ => None,
2796        })
2797        .collect();
2798
2799    let new_fns: BTreeMap<String, lex_ast::FnDecl> = BTreeMap::new();
2800    let report = lex_vcs::compute_diff(&old_fns, &new_fns, false);
2801    let empty_imports = lex_vcs::ImportMap::new();
2802
2803    match store.publish_program(&branch, &[], &report, &empty_imports, false) {
2804        Ok(outcome) => {
2805            // Remove the version record and archive, then update the index.
2806            let ver = record.version.clone();
2807            let _ = std::fs::remove_file(pkg_version_path(&state.root, name, &ver));
2808            let _ = std::fs::remove_file(pkg_archive_path(&state.root, name, &ver));
2809            // Update index: remove this version, set latest to previous if any.
2810            if let Some(mut idx) = load_pkg_index(&state.root, name) {
2811                idx.versions.retain(|v| v.version != ver);
2812                idx.latest = idx.versions.last().map(|v| v.version.clone());
2813                if idx.versions.is_empty() {
2814                    let _ = std::fs::remove_dir_all(pkg_name_dir(&state.root, name));
2815                } else {
2816                    let bytes = serde_json::to_vec_pretty(&idx).unwrap_or_default();
2817                    let _ = std::fs::write(pkg_index_path(&state.root, name), bytes);
2818                }
2819            }
2820            json_response(200, &serde_json::json!({
2821                "deleted": name,
2822                "version": ver,
2823                "ops": outcome.ops,
2824                "head_op": outcome.head_op,
2825            }))
2826        }
2827        Err(lex_store::StoreError::TypeError(errs)) => {
2828            error_with_detail(422, "type errors", serde_json::to_value(&errs).unwrap())
2829        }
2830        Err(e) => write_error_response("retract package", e),
2831    }
2832}
2833
2834#[cfg(test)]
2835mod dep_spec_tests {
2836    use super::DepSpec;
2837
2838    #[test]
2839    fn a_dual_spec_renders_both_git_and_vcs_refs_vcs_first() {
2840        let spec = DepSpec {
2841            registry: Some("vcs.lexlang.org/lex-official/lex-schema".into()),
2842            version: Some("^0.9".into()),
2843            git: Some("https://github.com/alpibrusl/lex-schema".into()),
2844            ..Default::default()
2845        };
2846        assert_eq!(
2847            spec.to_toml_inline().as_deref(),
2848            Some("{ registry = \"vcs.lexlang.org/lex-official/lex-schema\", version = \"^0.9\", git = \"https://github.com/alpibrusl/lex-schema\" }"),
2849        );
2850    }
2851
2852    #[test]
2853    fn bare_git_and_bare_registry_specs_render_their_own_keys() {
2854        let git = DepSpec { git: Some("https://x/g".into()), tag: Some("v1".into()), ..Default::default() };
2855        assert_eq!(git.to_toml_inline().as_deref(), Some("{ git = \"https://x/g\", tag = \"v1\" }"));
2856        let reg = DepSpec { registry: Some("vcs/r".into()), version: Some("1.0.0".into()), ..Default::default() };
2857        assert_eq!(reg.to_toml_inline().as_deref(), Some("{ registry = \"vcs/r\", version = \"1.0.0\" }"));
2858    }
2859
2860    #[test]
2861    fn an_empty_spec_renders_nothing() {
2862        assert_eq!(DepSpec::default().to_toml_inline(), None);
2863    }
2864}
2865
2866#[cfg(test)]
2867mod policy_ceiling_tests {
2868    use super::*;
2869    use lex_runtime::Policy;
2870    use std::path::PathBuf;
2871
2872    /// A maximally-permissive policy of the kind a malicious caller
2873    /// would put in a `/v1/run` body: every dangerous effect plus
2874    /// fs over `/`.
2875    fn permissive_request() -> Policy {
2876        Policy {
2877            allow_effects: ["io", "fs_read", "fs_write", "net", "proc"]
2878                .iter()
2879                .map(|s| s.to_string())
2880                .collect(),
2881            allow_fs_read: vec![PathBuf::from("/")],
2882            allow_fs_write: vec![PathBuf::from("/")],
2883            allow_net_host: Vec::new(),
2884            allow_proc: Vec::new(),
2885            allow_approval: Vec::new(),
2886            budget: None,
2887        }
2888    }
2889
2890    #[test]
2891    fn ceiling_drops_effects_the_caller_was_not_granted() {
2892        let ceiling = Policy {
2893            allow_effects: ["io", "time"].iter().map(|s| s.to_string()).collect(),
2894            ..Policy::default()
2895        };
2896        let got = clamp_policy(permissive_request(), &ceiling);
2897        assert!(got.allow_effects.contains("io"));
2898        assert!(!got.allow_effects.contains("proc"), "proc must not survive a ceiling without it");
2899        assert!(!got.allow_effects.contains("fs_write"));
2900        assert!(!got.allow_effects.contains("net"));
2901        // `time` is in the ceiling but not the request → intersection drops it.
2902        assert!(!got.allow_effects.contains("time"));
2903    }
2904
2905    #[test]
2906    fn ceiling_scopes_override_caller_scopes() {
2907        let ceiling = Policy {
2908            allow_effects: ["fs_read"].iter().map(|s| s.to_string()).collect(),
2909            allow_fs_read: vec![PathBuf::from("/srv/tenant")],
2910            ..Policy::default()
2911        };
2912        let got = clamp_policy(permissive_request(), &ceiling);
2913        // Caller asked for "/" but only the ceiling's scope survives —
2914        // an empty/wider caller list can never widen the ceiling.
2915        assert_eq!(got.allow_fs_read, vec![PathBuf::from("/srv/tenant")]);
2916        assert!(got.allow_fs_write.is_empty());
2917        assert!(got.allow_proc.is_empty());
2918        assert!(got.allow_net_host.is_empty());
2919    }
2920
2921    #[test]
2922    fn ceiling_caps_budget_and_prefers_the_smaller() {
2923        // Caller wants unlimited; ceiling caps it.
2924        let mut req = permissive_request();
2925        req.budget = None;
2926        let ceiling = Policy { budget: Some(1_000), ..Policy::default() };
2927        assert_eq!(clamp_policy(req, &ceiling).budget, Some(1_000));
2928
2929        // Caller asks for less than the ceiling → keep the caller's.
2930        let mut req2 = permissive_request();
2931        req2.budget = Some(50);
2932        let ceiling2 = Policy { budget: Some(1_000), ..Policy::default() };
2933        assert_eq!(clamp_policy(req2, &ceiling2).budget, Some(50));
2934    }
2935
2936    #[test]
2937    fn empty_ceiling_is_pure_only() {
2938        let got = clamp_policy(permissive_request(), &Policy::default());
2939        assert!(got.allow_effects.is_empty(), "an empty ceiling grants nothing");
2940        assert!(got.allow_proc.is_empty());
2941        assert!(got.allow_fs_write.is_empty());
2942    }
2943}
2944
2945#[cfg(test)]
2946mod public_read_tests {
2947    use super::*;
2948
2949    /// Write a minimal package (index + per-version record + an archive
2950    /// blob) straight into a temp store, bypassing the publish pipeline.
2951    fn seed_pkg(root: &std::path::Path, name: &str, version: &str) {
2952        let record = PkgRecord {
2953            name: name.to_string(),
2954            version: version.to_string(),
2955            head_op: Some(format!("op-{name}")),
2956            published_at: 1,
2957            function_names: vec![format!("{name}.f")],
2958            dependencies: vec![],
2959            dependency_specs: Default::default(),
2960            ops: vec![],
2961        };
2962        save_pkg_record(root, &record, Some(format!("ARCHIVE:{name}@{version}").as_bytes()))
2963            .expect("seed package");
2964    }
2965
2966    #[test]
2967    fn new_package_defaults_to_private() {
2968        let tmp = tempfile::TempDir::new().unwrap();
2969        seed_pkg(tmp.path(), "lex-schema", "0.9.2");
2970        assert!(!pkg_is_public(tmp.path(), "lex-schema"));
2971        // Unknown packages are also "not public" — never distinguished.
2972        assert!(!pkg_is_public(tmp.path(), "does-not-exist"));
2973    }
2974
2975    #[test]
2976    fn set_visibility_round_trips_and_index_persists() {
2977        let tmp = tempfile::TempDir::new().unwrap();
2978        let state = State::open(tmp.path().to_path_buf()).unwrap();
2979        seed_pkg(tmp.path(), "lex-schema", "0.9.2");
2980
2981        let _ = pkg_set_visibility_handler(&state, "lex-schema", r#"{"visibility":"public"}"#);
2982        assert!(pkg_is_public(tmp.path(), "lex-schema"));
2983        // The version list survives the index rewrite (we don't clobber it).
2984        let idx = load_pkg_index(tmp.path(), "lex-schema").unwrap();
2985        assert_eq!(idx.latest.as_deref(), Some("0.9.2"));
2986        assert_eq!(idx.versions.len(), 1);
2987
2988        let _ = pkg_set_visibility_handler(&state, "lex-schema", r#"{"visibility":"private"}"#);
2989        assert!(!pkg_is_public(tmp.path(), "lex-schema"));
2990    }
2991
2992    #[test]
2993    fn set_visibility_on_unknown_package_is_a_noop() {
2994        let tmp = tempfile::TempDir::new().unwrap();
2995        let state = State::open(tmp.path().to_path_buf()).unwrap();
2996        // No package seeded → handler returns 404 and writes nothing.
2997        let _ = pkg_set_visibility_handler(&state, "ghost", r#"{"visibility":"public"}"#);
2998        assert!(load_pkg_index(tmp.path(), "ghost").is_none());
2999    }
3000
3001    #[test]
3002    fn public_listing_omits_private_packages() {
3003        let tmp = tempfile::TempDir::new().unwrap();
3004        let state = State::open(tmp.path().to_path_buf()).unwrap();
3005        seed_pkg(tmp.path(), "pub-pkg", "1.0.0");
3006        seed_pkg(tmp.path(), "priv-pkg", "1.0.0");
3007        let _ = pkg_set_visibility_handler(&state, "pub-pkg", r#"{"visibility":"public"}"#);
3008
3009        let names = public_pkg_names(tmp.path());
3010        assert_eq!(names, vec!["pub-pkg".to_string()]);
3011    }
3012
3013    #[test]
3014    fn resolve_public_maps_routes() {
3015        let get = Method::Get;
3016        assert_eq!(resolve_public(&get, "").unwrap(), PublicTarget::List);
3017        assert_eq!(resolve_public(&get, "/").unwrap(), PublicTarget::List);
3018        assert_eq!(
3019            resolve_public(&get, "/lex-schema").unwrap(),
3020            PublicTarget::Latest("lex-schema".into())
3021        );
3022        assert_eq!(
3023            resolve_public(&get, "/lex-schema/versions").unwrap(),
3024            PublicTarget::Versions("lex-schema".into())
3025        );
3026        assert_eq!(
3027            resolve_public(&get, "/lex-schema/head").unwrap(),
3028            PublicTarget::Head("lex-schema".into())
3029        );
3030        assert_eq!(
3031            resolve_public(&get, "/lex-schema/0.9.2").unwrap(),
3032            PublicTarget::Version("lex-schema".into(), "0.9.2".into())
3033        );
3034        assert_eq!(
3035            resolve_public(&get, "/lex-schema/0.9.2/archive").unwrap(),
3036            PublicTarget::Archive("lex-schema".into(), "0.9.2".into())
3037        );
3038    }
3039
3040    #[test]
3041    fn resolve_public_rejects_bad_method_and_traversal() {
3042        // Non-GET → 405.
3043        assert_eq!(resolve_public(&Method::Put, "/lex-schema"), Err(405));
3044        assert_eq!(resolve_public(&Method::Post, "").err(), Some(405));
3045        // Path traversal / invalid segments → 404, never a filesystem touch.
3046        assert_eq!(resolve_public(&Method::Get, "/.."), Err(404));
3047        assert_eq!(resolve_public(&Method::Get, "/lex-schema/../etc"), Err(404));
3048        assert_eq!(resolve_public(&Method::Get, "/a/b/c/d"), Err(404));
3049        // Slashes elsewhere can't smuggle a deep path: each segment is checked.
3050        assert!(resolve_public(&Method::Get, "/lex schema").is_err());
3051    }
3052}