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