Skip to main content

lex_store/
store.rs

1//! `Store` — content-addressed code repository.
2//!
3//! The filesystem is the source of truth. All operations read/write JSON
4//! files under `<root>/stages/<SigId>/`. There is no SQLite cache: every
5//! query walks the directory and parses what's needed. `cargo test`
6//! runs aren't perf-critical and the §4.6 acceptance requires the
7//! rebuild-from-filesystem property anyway.
8
9use crate::branches::DEFAULT_BRANCH;
10use crate::model::*;
11use lex_ast::{sig_id, stage_id, Stage};
12use serde::de::DeserializeOwned;
13use serde::Serialize;
14use std::collections::BTreeMap;
15use std::fs;
16use std::path::{Path, PathBuf};
17use std::time::{SystemTime, UNIX_EPOCH};
18
19#[derive(Debug, thiserror::Error)]
20pub enum StoreError {
21    #[error("io error: {0}")]
22    Io(#[from] std::io::Error),
23    #[error("serialization error: {0}")]
24    Serde(#[from] serde_json::Error),
25    #[error("imports cannot be published as stages")]
26    CannotPublishImport,
27    #[error("unknown stage_id `{0}`")]
28    UnknownStage(String),
29    #[error("unknown sig_id `{0}`")]
30    UnknownSig(String),
31    #[error("invalid lifecycle transition: {0}")]
32    InvalidTransition(String),
33    #[error("unknown branch `{0}`")]
34    UnknownBranch(String),
35    /// A branch-head advance (e.g. the ref half of `op push`) was
36    /// asked to move `branch` to `attempted`, but `attempted` is not a
37    /// descendant of the branch's `current` head — a non-fast-forward
38    /// that would orphan history. Refused, git-style, so a disjoint or
39    /// diverged push can't silently clobber a shared branch. The op
40    /// objects may already be present; only the ref is left unchanged.
41    #[error("non-fast-forward on `{branch}`: {attempted} is not a descendant of current head {current}")]
42    NonFastForward { branch: String, current: lex_vcs::OpId, attempted: lex_vcs::OpId },
43    #[error("unknown blob `{0}`")]
44    UnknownBlob(String),
45    #[error("unknown blob ref `{namespace}/{key}`")]
46    UnknownBlobRef { namespace: String, key: String },
47    #[error("unknown op_id `{0}`")]
48    UnknownOp(lex_vcs::OpId),
49    /// A typed AST transform (#280) — e.g. `ReplaceMatchArm` — was
50    /// asked to operate on a node it couldn't address (wrong kind,
51    /// out-of-range arm index, unknown NodeId, etc.). Distinct from
52    /// `TypeError` (which means the transform succeeded but its
53    /// output didn't typecheck) so callers can render the right
54    /// error message.
55    #[error("transform failed: {0}")]
56    TransformError(lex_ast::TransformError),
57    #[error(transparent)]
58    Apply(#[from] lex_vcs::ApplyError),
59    /// The candidate program — i.e. the source the caller is
60    /// publishing — doesn't typecheck. The branch head is unchanged
61    /// and no op records are persisted. Issue #130's "always-valid
62    /// HEAD" invariant: the gate runs before any side effect, so a
63    /// type-broken publish leaves no footprint.
64    #[error("type errors in published program: {} error(s)", .0.len())]
65    TypeError(Vec<lex_types::TypeError>),
66    /// A dependency being resolved for the write-time gate (#930) is a
67    /// multi-module package. Per-module signature extraction (picking the
68    /// imported module's file out of the de-flattened tree) is not yet
69    /// implemented; single-module (leaf) dependencies resolve today. A
70    /// caller can treat this as "cannot resolve here" rather than a hard
71    /// failure.
72    #[error("multi-module dependency resolution is not yet supported")]
73    UnsupportedMultiModuleDependency,
74    /// A typed issue's example targets a function that isn't declared at
75    /// the head being evaluated (#949) — the example cannot be attached, so
76    /// the issue cannot be judged there.
77    #[error("issue example targets `{0}`, which is not declared at this head")]
78    IssueTarget(String),
79    /// The op was persisted but a `required_attestations` rule in
80    /// `policy.json` (#245) refused to advance the branch head past
81    /// it. The op record is durable — re-running with the missing
82    /// attestations recorded will succeed without re-persisting —
83    /// but the branch is unchanged. Surfaced as a structured JSON
84    /// envelope on the HTTP API.
85    #[error(
86        "branch advance blocked: op {} missing attestations: {}",
87        .0.op_id, .0.missing.join(", ")
88    )]
89    BranchAdvanceBlocked(crate::policy::BranchAdvanceBlocked),
90    /// All retry attempts of the CAS branch-head advance failed
91    /// because another writer kept advancing the same branch
92    /// (#262). The op record itself is durable in the op log
93    /// (orphaned), so re-running with backoff would eventually
94    /// land — return `503 Contention { retry_after }` from the
95    /// HTTP API and let the client back off.
96    #[error("branch advance contention on `{branch}`: {attempts} retries exhausted")]
97    Contention { branch: String, attempts: u32 },
98    /// The op was persisted but its stage carries an attestation
99    /// produced by a retroactively quarantined tool (#248). The
100    /// branch head is unchanged. The op record stays in the log
101    /// (audit trail intact); re-running with the producer
102    /// unblocked, or with un-contaminated attestations, succeeds
103    /// without re-persisting the op.
104    #[error(
105        "branch advance blocked: op {} touches stage {} with an attestation from \
106         quarantined producer `{}` (blocked at {}, attestation at {})",
107        .0.op_id, .0.stage_id, .0.tool_id, .0.blocked_at, .0.attestation_at
108    )]
109    ProducerBlocked(crate::policy::ProducerBlocked),
110    /// The op would push its session's monotonic budget over the
111    /// cap configured in `policy.session_budgets` (#292 slice 3).
112    /// The op is *not* persisted; the branch head is unchanged.
113    /// The caller should either start a new session, raise the
114    /// cap, or refactor to fit the budget. HTTP API maps to 503.
115    #[error("session `{session_id}` budget exceeded: spent_after={spent_after} > cap={cap}")]
116    BudgetExceeded {
117        session_id: String,
118        cap: u64,
119        spent_after: u64,
120    },
121}
122
123/// The outcome returned by [`Store::publish_program`].
124#[derive(Debug, Clone, serde::Serialize)]
125pub struct PublishOutcome {
126    pub ops: Vec<PublishOp>,
127    pub head_op: Option<lex_vcs::OpId>,
128}
129
130/// Everything a regenerator needs to *replay* an op (#836 G3), produced
131/// by [`Store::replay_request`]. The model call is external: a harness
132/// feeds `prompt` + `parent_program` to `model`, then hands the
133/// regenerated stage to [`Store::replay_compare`].
134#[derive(Debug, Clone, serde::Serialize)]
135pub struct ReplayRequest {
136    pub op_id: String,
137    /// The sig the op changed — the function to regenerate.
138    pub target_sig: String,
139    /// The target function's name (the recorded stage is a function
140    /// for every replayable op).
141    #[serde(default, skip_serializing_if = "Option::is_none")]
142    pub target_name: Option<String>,
143    /// The target function's rendered signature (`fn name(...) -> T`),
144    /// so a regenerator knows the interface to implement without
145    /// re-deriving it from the sig hash.
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    pub target_signature: Option<String>,
148    /// The stage id a faithful regeneration should reproduce.
149    pub expected_stage_id: String,
150    /// The recorded intent prompt (`None` if the op carried no intent).
151    pub prompt: Option<String>,
152    /// The recorded model (`provider/name[@version]`), if any.
153    pub model: Option<String>,
154    /// The recorded session id, if any.
155    pub session_id: Option<String>,
156    /// The program the change was made against — the parent state
157    /// rendered to source — the context a regenerator needs.
158    pub parent_program: String,
159}
160
161/// The result of comparing a regenerated candidate against an op's
162/// recorded output (#836 G3), returned by [`Store::replay_compare`]
163/// after it emits the `Replay` attestation.
164#[derive(Debug, Clone, serde::Serialize)]
165pub struct ReplayOutcome {
166    pub op_id: String,
167    pub expected_stage_id: String,
168    /// The candidate's stage id when it regenerated the same sig, else
169    /// `None`.
170    pub produced_stage_id: Option<String>,
171    /// Whether the regeneration reproduced the recorded change (exact or
172    /// behavioral).
173    pub reproduced: bool,
174    /// Set when reproduction was behavioral (same values over N sampled
175    /// inputs) rather than an exact stage-id match. `None` for an exact
176    /// match or a genuine miss.
177    #[serde(default, skip_serializing_if = "Option::is_none")]
178    pub behavioral_samples: Option<usize>,
179    /// The id of the `Replay` attestation this comparison emitted.
180    pub attestation_id: String,
181}
182
183/// One applied operation within a [`PublishOutcome`].
184#[derive(Debug, Clone, serde::Serialize)]
185pub struct PublishOp {
186    pub op_id: lex_vcs::OpId,
187    pub kind: serde_json::Value,
188}
189
190/// One entry in the per-`SigId` stage history surfaced by
191/// `Store::sig_history`. Newest-first ordering is the responsibility
192/// of the producer.
193#[derive(Debug, Clone, serde::Serialize, PartialEq)]
194pub struct StageHistoryEntry {
195    pub stage_id: String,
196    pub status: StageStatus,
197    /// Wall-clock seconds of the most recent transition.
198    pub last_at: u64,
199    /// Wall-clock seconds when this stage was first written to the
200    /// store (its initial Draft transition). `None` for stages
201    /// whose lifecycle log doesn't include an explicit Draft entry
202    /// — shouldn't happen for stages published via `Store::publish`,
203    /// but the type allows hand-edited stores.
204    #[serde(skip_serializing_if = "Option::is_none")]
205    pub published_at: Option<u64>,
206}
207
208/// Per-candidate metadata surfaced by [`Store::list_candidates`]
209/// (#294). Returned sorted by `op_id` for deterministic output.
210#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
211pub struct CandidateInfo {
212    pub op_id: lex_vcs::OpId,
213    pub stage_id: lex_vcs::StageId,
214    /// Author intent. Always set for `Candidate` ops emitted via
215    /// [`Store::propose_candidate`]; `None` only if a
216    /// hand-written raw op skipped the intent tag.
217    pub intent_id: Option<lex_vcs::IntentId>,
218}
219
220/// One line of `stage_index.jsonl`. See `Store::lookup_lifecycle`.
221#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
222struct StageIndexEntry {
223    stage_id: String,
224    sig_id: String,
225}
226
227/// Sentinel `sig_id` value recording "a full scan already established
228/// this stage_id exists nowhere in the store" (#825). Never a real
229/// sig — sig directory names are never empty.
230const MISSING_STAGE_MARKER: &str = "";
231
232/// Resolves a head's external (registry/git) dependencies to their public
233/// module signatures, so the write-time gate can type-check a head that keeps
234/// `import "<pkg>/mod" as <alias>` edges instead of inlining the dependency
235/// (#930). The returned map is keyed by import *reference* (`"lex-nt/lib"`) →
236/// that dependency module's record type (as [`crate::render::module_record_at_op`]
237/// or a source-based equivalent produces), exactly the shape
238/// [`lex_types::check_program_with_modules`] consumes.
239///
240/// Implementations differ by context and live in the crate that has the
241/// resolution machinery: the client (`lex publish`) resolves from the
242/// working-copy `lex.lock` + local package cache; the hub resolves from the
243/// committed lock + its own hosted stores (cross-tenant), which a single
244/// [`Store`] cannot reach on its own. When no resolver is installed the gate
245/// resolves an empty map — only stdlib binds and any external reference is an
246/// unbound-name error, exactly as before #930 (so inlined heads, which carry
247/// no external edges, are unaffected).
248pub trait DepResolver: Send + Sync {
249    /// `head_op` is the op the head is known by when the gate has one (merge,
250    /// patch and hub-verify reconstruct a committed head); `None` for a
251    /// candidate not yet committed (`publish`), where the client resolver
252    /// falls back to the working-copy lock.
253    fn resolve_modules(
254        &self,
255        stages: &[Stage],
256        head_op: Option<&str>,
257    ) -> BTreeMap<String, lex_types::Ty>;
258}
259
260pub struct Store {
261    root: PathBuf,
262    /// Optional dependency resolver (#930). Injected by the client or the hub;
263    /// `None` in a bare store means the gate resolves an empty module map.
264    resolver: Option<std::sync::Arc<dyn DepResolver>>,
265}
266
267impl Store {
268    /// Open or create a store rooted at `root`.
269    pub fn open(root: impl AsRef<Path>) -> Result<Self, StoreError> {
270        let root = root.as_ref().to_path_buf();
271        fs::create_dir_all(root.join("stages"))?;
272        fs::create_dir_all(root.join("traces"))?;
273        let store = Self { root, resolver: None };
274        store.ensure_stage_index();
275        Ok(store)
276    }
277
278    /// Install the dependency resolver the write-time gate uses to type-check
279    /// heads that keep external `import` edges (#930). Builder-style so a
280    /// caller can write `Store::open(p)?.with_dep_resolver(r)`.
281    pub fn with_dep_resolver(mut self, resolver: std::sync::Arc<dyn DepResolver>) -> Self {
282        self.resolver = Some(resolver);
283        self
284    }
285
286    /// Set the dependency resolver in place (for a `Store` already owned, e.g.
287    /// behind a `Mutex` in the HTTP `State`).
288    pub fn set_dep_resolver(&mut self, resolver: std::sync::Arc<dyn DepResolver>) {
289        self.resolver = Some(resolver);
290    }
291
292    /// The resolved dependency module map for a head being gated — the
293    /// installed resolver's answer, or empty when none is installed (#930).
294    fn resolved_modules(
295        &self,
296        stages: &[Stage],
297        head_op: Option<&str>,
298    ) -> BTreeMap<String, lex_types::Ty> {
299        match &self.resolver {
300            Some(r) => r.resolve_modules(stages, head_op),
301            None => BTreeMap::new(),
302        }
303    }
304
305    /// One-time migration for a store that predates the reverse
306    /// index (#822), or whose previous rebuild pass never finished
307    /// (e.g. the process was killed or its client disconnected
308    /// mid-request — server-side work keeps running either way, but
309    /// a *restart* genuinely stops it): build the index in a single
310    /// pass instead of leaving every subsequent `lookup_lifecycle`
311    /// call to discover its own entry via the slow per-call scan-
312    /// and-backfill fallback.
313    ///
314    /// That per-call fallback is fine for the rare individual miss
315    /// it was designed for, but pathological as a *bulk* cold-start
316    /// strategy: on a tenant with a few thousand functions it means
317    /// redoing an O(total sigs) scan from scratch for *each* of a
318    /// few thousand cold entries — O(total sigs²) — which measured
319    /// as a near-stall (page-cache thrashing) on a memory-
320    /// constrained host. A single pass over `list_sigs()` is
321    /// O(total sigs) total.
322    ///
323    /// Gated on a dedicated completion marker
324    /// (`stage_index.complete`), NOT on `stage_index.jsonl`'s mere
325    /// existence — a partially-built index file (left behind by an
326    /// interrupted rebuild, lazy or bulk) must still trigger a
327    /// re-run so the remaining entries get backfilled in one more
328    /// cheap O(total sigs) pass, not silently be mistaken for
329    /// "already done" and fall back to the slow per-call path for
330    /// whatever's left. `rebuild_stage_index` already skips entries
331    /// it finds present, so re-running it against a partial index
332    /// only does the work that remains. The marker is written only
333    /// after a full pass returns `Ok`, so a failed pass (e.g. an I/O
334    /// error partway through `list_sigs`) is retried on the next
335    /// open rather than being marked done.
336    ///
337    /// Runs once per `Store::open` call — which, in a long-lived
338    /// server (lex-hub caches one `Store` per tenant for the life of
339    /// the process), means once per tenant per process lifetime, not
340    /// once per request. Once the marker exists (the steady state
341    /// after the first successful run on any given host) this is a
342    /// single cheap file-existence check. Best-effort like the rest
343    /// of the index: any failure here just leaves the slower per-call
344    /// fallback as the only path, never breaks correctness.
345    fn ensure_stage_index(&self) {
346        if self.stage_index_complete_marker_path().exists() {
347            return;
348        }
349        if self.rebuild_stage_index().is_ok() {
350            let _ = fs::write(self.stage_index_complete_marker_path(), "");
351        }
352    }
353
354    fn stage_index_complete_marker_path(&self) -> PathBuf {
355        self.root.join("stage_index.complete")
356    }
357
358    /// Build (or top up) the reverse index in one pass over every
359    /// SigId in the store, rather than relying on `lookup_lifecycle`
360    /// to discover entries one at a time. Safe to call at any time,
361    /// including on a partially-built index (e.g. one left behind by
362    /// an interrupted request that was populating it lazily): already-
363    /// indexed stage_ids are skipped, so this only does the work that
364    /// remains. Returns the number of newly-added entries.
365    pub fn rebuild_stage_index(&self) -> Result<usize, StoreError> {
366        // A sig's lifecycle can list the same stage_id more than once
367        // (Draft, then later Active, then Deprecated all carry the
368        // same stage_id with a different status) -- track newly-seen
369        // keys locally too, not just what was already on disk at the
370        // start, so a repeated stage_id within one sig's transitions
371        // doesn't get appended to the index more than once.
372        let mut existing = self.load_stage_index();
373        let mut added = 0usize;
374        for sig in self.list_sigs()? {
375            let Ok(life) = self.read_lifecycle(&sig) else { continue };
376            for t in &life.transitions {
377                if !existing.contains_key(&t.stage_id) {
378                    self.append_stage_index_entry(&t.stage_id, &sig);
379                    existing.insert(t.stage_id.clone(), sig.clone());
380                    added += 1;
381                }
382            }
383        }
384        Ok(added)
385    }
386
387    pub fn root(&self) -> &Path {
388        &self.root
389    }
390
391    // ── Generic content-addressed blobs (#5 / M6.1) ──────────────────────────
392    //
393    // The stage store holds typed Lex ASTs; loom-style artifacts (generated
394    // code, JSON, prose) are opaque text. These blob methods give the store a
395    // generic content-addressed object alongside stages, plus a lightweight
396    // ref namespace so callers can bind names (e.g. a sprint's node ids) to
397    // blob shas without touching the operation-log branch machinery.
398    //
399    // The sha is the lowercase hex SHA-256 of the content's UTF-8 bytes —
400    // identical to Lex's `crypto.sha256_str`, so a blob written here and an
401    // artifact content-addressed in loom's SQLite store share the same id and
402    // are interchangeable by reference. Store-scoped, so under lex-hub each
403    // tenant store gets its own blob space for free.
404
405    fn blobs_dir(&self) -> PathBuf {
406        self.root.join("blobs")
407    }
408
409    fn blob_refs_dir(&self) -> PathBuf {
410        self.root.join("blobrefs")
411    }
412
413    /// Content-address `content` and persist it under `<root>/blobs/<sha>`.
414    /// Returns the sha. Idempotent: re-putting identical content is a no-op.
415    /// Concurrency-safe — writes to a unique temp file then atomically renames
416    /// onto the content-addressed path, so parallel writers of the same content
417    /// can't corrupt it.
418    pub fn put_blob(&self, content: &str) -> Result<String, StoreError> {
419        use sha2::{Digest, Sha256};
420        let sha = hex::encode(Sha256::digest(content.as_bytes()));
421        let dir = self.blobs_dir();
422        let path = dir.join(&sha);
423        if path.exists() {
424            return Ok(sha);
425        }
426        fs::create_dir_all(&dir)?;
427        static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
428        let n = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
429        let tmp = dir.join(format!(".{sha}.{}.{n}.tmp", std::process::id()));
430        fs::write(&tmp, content.as_bytes())?;
431        // rename is atomic on the same filesystem; identical content makes a
432        // last-writer-wins race harmless.
433        fs::rename(&tmp, &path)?;
434        Ok(sha)
435    }
436
437    /// Read a blob by its sha. `UnknownBlob` if absent.
438    pub fn get_blob(&self, sha: &str) -> Result<String, StoreError> {
439        match fs::read_to_string(self.blobs_dir().join(sha)) {
440            Ok(s) => Ok(s),
441            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
442                Err(StoreError::UnknownBlob(sha.to_string()))
443            }
444            Err(e) => Err(StoreError::Io(e)),
445        }
446    }
447
448    /// Whether a blob with this sha exists.
449    pub fn has_blob(&self, sha: &str) -> bool {
450        self.blobs_dir().join(sha).exists()
451    }
452
453    /// Bind `key` to a blob `sha` within `namespace` (e.g. namespace
454    /// `"loom/sprint-abc"`, key `"build-node"`). Overwrites an existing
455    /// binding. The namespace may contain `/`; neither namespace nor key may
456    /// contain a `..` path component.
457    pub fn set_blob_ref(&self, namespace: &str, key: &str, sha: &str) -> Result<(), StoreError> {
458        let dir = self.blob_ref_namespace_dir(namespace, key)?;
459        fs::create_dir_all(&dir)?;
460        fs::write(dir.join(key), sha.as_bytes())?;
461        Ok(())
462    }
463
464    /// Resolve `namespace`/`key` to a blob sha. `UnknownBlobRef` if unbound.
465    pub fn get_blob_ref(&self, namespace: &str, key: &str) -> Result<String, StoreError> {
466        let dir = self.blob_ref_namespace_dir(namespace, key)?;
467        match fs::read_to_string(dir.join(key)) {
468            Ok(s) => Ok(s.trim().to_string()),
469            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Err(StoreError::UnknownBlobRef {
470                namespace: namespace.to_string(),
471                key: key.to_string(),
472            }),
473            Err(e) => Err(StoreError::Io(e)),
474        }
475    }
476
477    /// Blob-ref namespace for committed lockfiles (#930 phase 2b-1).
478    const LOCK_NS: &'static str = "lock";
479
480    /// Record the `lex.lock` committed with the package head `head_op` — the
481    /// exact dependency versions and op-log heads that head was built and
482    /// type-checks against (#930 phase 2b-1: "HEAD + its committed lex.lock
483    /// always type-checks"). Content-addressed via [`Self::put_blob`] and
484    /// bound under the `lock` namespace keyed by the head op, so it is
485    /// idempotent (a re-push converges) and travels with the package through
486    /// the same object-sync path as stages and intents. Keyed by head op
487    /// rather than by branch so re-verifying a *historical* head resolves it
488    /// against the lock that head actually committed, not whatever the branch
489    /// points at now.
490    pub fn set_committed_lock(&self, head_op: &str, lock_toml: &str) -> Result<(), StoreError> {
491        let sha = self.put_blob(lock_toml)?;
492        self.set_blob_ref(Self::LOCK_NS, head_op, &sha)
493    }
494
495    /// The `lex.lock` committed with `head_op`, or `None` when the head
496    /// carries no committed lock — a dependency-free package, or one
497    /// published before locks were committed (the write-time gate then has no
498    /// registry/git dependencies to resolve, exactly as today).
499    pub fn committed_lock(&self, head_op: &str) -> Result<Option<String>, StoreError> {
500        match self.get_blob_ref(Self::LOCK_NS, head_op) {
501            Ok(sha) => Ok(Some(self.get_blob(&sha)?)),
502            Err(StoreError::UnknownBlobRef { .. }) => Ok(None),
503            Err(e) => Err(e),
504        }
505    }
506
507    /// All `key → sha` bindings in a namespace (e.g. every artifact in a
508    /// sprint). Empty map if the namespace has no bindings yet.
509    pub fn list_blob_refs(
510        &self,
511        namespace: &str,
512    ) -> Result<std::collections::BTreeMap<String, String>, StoreError> {
513        let dir = self.blob_ref_namespace_dir(namespace, "x")?;
514        let mut out = std::collections::BTreeMap::new();
515        let entries = match fs::read_dir(&dir) {
516            Ok(e) => e,
517            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out),
518            Err(e) => return Err(StoreError::Io(e)),
519        };
520        for entry in entries {
521            let entry = entry?;
522            if entry.file_type()?.is_file() {
523                let key = entry.file_name().to_string_lossy().to_string();
524                let sha = fs::read_to_string(entry.path())?.trim().to_string();
525                out.insert(key, sha);
526            }
527        }
528        Ok(out)
529    }
530
531    // Resolve the on-disk dir for a (namespace, key), rejecting `..` traversal
532    // and `/` in the key. `key` is validated but not joined here (callers join
533    // it themselves so `list_blob_refs` can pass a dummy).
534    fn blob_ref_namespace_dir(&self, namespace: &str, key: &str) -> Result<PathBuf, StoreError> {
535        if key.contains('/') || key.contains('\\') || key.split('/').any(|c| c == "..") {
536            return Err(StoreError::UnknownBlobRef {
537                namespace: namespace.to_string(),
538                key: key.to_string(),
539            });
540        }
541        let mut dir = self.blob_refs_dir();
542        for comp in namespace.split('/') {
543            if comp == ".." || comp.contains('\\') {
544                return Err(StoreError::UnknownBlobRef {
545                    namespace: namespace.to_string(),
546                    key: key.to_string(),
547                });
548            }
549            if !comp.is_empty() {
550                dir.push(comp);
551            }
552        }
553        Ok(dir)
554    }
555
556    fn now() -> u64 {
557        SystemTime::now()
558            .duration_since(UNIX_EPOCH)
559            .map(|d| d.as_secs())
560            .unwrap_or(0)
561    }
562
563    fn sig_dir(&self, sig: &str) -> PathBuf {
564        self.root.join("stages").join(sig)
565    }
566    fn impl_dir(&self, sig: &str) -> PathBuf {
567        self.sig_dir(sig).join("implementations")
568    }
569    fn tests_dir(&self, sig: &str) -> PathBuf {
570        self.sig_dir(sig).join("tests")
571    }
572    fn specs_dir(&self, sig: &str) -> PathBuf {
573        self.sig_dir(sig).join("specs")
574    }
575    fn lifecycle_path(&self, sig: &str) -> PathBuf {
576        self.sig_dir(sig).join("lifecycle.json")
577    }
578
579    // ---- publish ----
580
581    /// Publish a stage as **Draft**. Returns the StageId.
582    /// Idempotent: republishing the same canonical AST returns the same
583    /// StageId without writing duplicates.
584    pub fn publish(&self, stage: &Stage) -> Result<String, StoreError> {
585        self.publish_signed(stage, None)
586    }
587
588    /// Like [`Self::publish`] but optionally attaches an Ed25519
589    /// signature over the StageId (#227). When `signer` is `Some`,
590    /// the persisted metadata gets a `signature` field that
591    /// downstream consumers can verify via
592    /// [`lex_vcs::verify_stage_id`].
593    ///
594    /// Idempotency: if a metadata file already exists the signature
595    /// is *not* re-written. This preserves "republishing is a no-op"
596    /// even across different signers — promoting a signed stage
597    /// requires a fresh stage hash anyway, so a metadata overwrite
598    /// would be the wrong primitive.
599    pub fn publish_signed(
600        &self,
601        stage: &Stage,
602        signer: Option<&lex_vcs::Keypair>,
603    ) -> Result<String, StoreError> {
604        let sig = sig_id(stage).ok_or(StoreError::CannotPublishImport)?;
605        let stage_id = stage_id(stage).ok_or(StoreError::CannotPublishImport)?;
606        let name = stage_name(stage).to_string();
607
608        fs::create_dir_all(self.impl_dir(&sig))?;
609        fs::create_dir_all(self.tests_dir(&sig))?;
610        fs::create_dir_all(self.specs_dir(&sig))?;
611
612        let ast_path = self.impl_dir(&sig).join(format!("{}.ast.json", stage_id));
613        let delta_path = self.impl_dir(&sig).join(format!("{}.delta.json", stage_id));
614        let meta_path = self
615            .impl_dir(&sig)
616            .join(format!("{}.metadata.json", stage_id));
617
618        // #261 slice 3: try delta encoding against the most recent
619        // prior stage in this sig's lifecycle. Falls back to a full
620        // snapshot when (a) no prior stage exists, (b) the diff
621        // ratio is over the threshold, or (c) the delta chain is
622        // already at its cap. The decision is internal — callers
623        // see the same `Stage` object on `get_ast` regardless.
624        if !ast_path.exists() && !delta_path.exists() {
625            self.persist_stage_bytes(&sig, &stage_id, stage, &ast_path, &delta_path)?;
626        }
627        if !meta_path.exists() {
628            let signature = signer.map(|kp| kp.sign_stage_id(&stage_id));
629            let metadata = Metadata {
630                stage_id: stage_id.clone(),
631                sig_id: sig.clone(),
632                name,
633                published_at: Self::now(),
634                note: None,
635                signature,
636            };
637            write_canonical_json(&meta_path, &metadata)?;
638        }
639
640        // Lifecycle: append a Draft transition for first publish.
641        let mut life = self.read_lifecycle(&sig).unwrap_or_else(|_| Lifecycle {
642            sig_id: sig.clone(),
643            ..Default::default()
644        });
645        if !life.transitions.iter().any(|t| t.stage_id == stage_id) {
646            life.transitions.push(Transition {
647                stage_id: stage_id.clone(),
648                from: StageStatus::Draft, // synthesized; "from" of first transition is itself
649                to: StageStatus::Draft,
650                at: Self::now(),
651                reason: None,
652            });
653            self.write_lifecycle(&sig, &life)?;
654            // Register the new stage_id's owning sig up front so a
655            // later `lookup_lifecycle` (e.g. `get_ast`) never needs
656            // to fall back to a full tenant-wide scan for it.
657            self.append_stage_index_entry(&stage_id, &sig);
658        }
659        Ok(stage_id)
660    }
661
662    // ---- lifecycle ----
663
664    pub fn activate(&self, stage_id: &str) -> Result<(), StoreError> {
665        let (sig, mut life) = self.lookup_lifecycle(stage_id)?;
666        // Demote any currently-Active impls for this SigId to Deprecated.
667        let active = life.current_active().map(|s| s.to_string());
668        if let Some(prev) = active {
669            if prev != stage_id {
670                life.transitions.push(Transition {
671                    stage_id: prev,
672                    from: StageStatus::Active,
673                    to: StageStatus::Deprecated,
674                    at: Self::now(),
675                    reason: Some("superseded".into()),
676                });
677            }
678        }
679        let cur = life.status_of(stage_id);
680        if cur == Some(StageStatus::Tombstone) {
681            return Err(StoreError::InvalidTransition(
682                "tombstoned cannot be activated".into(),
683            ));
684        }
685        life.transitions.push(Transition {
686            stage_id: stage_id.into(),
687            from: cur.unwrap_or(StageStatus::Draft),
688            to: StageStatus::Active,
689            at: Self::now(),
690            reason: None,
691        });
692        self.write_lifecycle(&sig, &life)
693    }
694
695    pub fn deprecate(&self, stage_id: &str, reason: impl Into<String>) -> Result<(), StoreError> {
696        let (sig, mut life) = self.lookup_lifecycle(stage_id)?;
697        let cur = life
698            .status_of(stage_id)
699            .ok_or_else(|| StoreError::UnknownStage(stage_id.into()))?;
700        if cur != StageStatus::Active {
701            return Err(StoreError::InvalidTransition(format!(
702                "{cur:?} ⇒ Deprecated"
703            )));
704        }
705        life.transitions.push(Transition {
706            stage_id: stage_id.into(),
707            from: cur,
708            to: StageStatus::Deprecated,
709            at: Self::now(),
710            reason: Some(reason.into()),
711        });
712        self.write_lifecycle(&sig, &life)
713    }
714
715    pub fn tombstone(&self, stage_id: &str) -> Result<(), StoreError> {
716        let (sig, mut life) = self.lookup_lifecycle(stage_id)?;
717        let cur = life
718            .status_of(stage_id)
719            .ok_or_else(|| StoreError::UnknownStage(stage_id.into()))?;
720        if cur != StageStatus::Deprecated {
721            return Err(StoreError::InvalidTransition(format!(
722                "{cur:?} ⇒ Tombstone"
723            )));
724        }
725        life.transitions.push(Transition {
726            stage_id: stage_id.into(),
727            from: cur,
728            to: StageStatus::Tombstone,
729            at: Self::now(),
730            reason: None,
731        });
732        self.write_lifecycle(&sig, &life)
733    }
734
735    // ---- queries ----
736
737    /// The current Active StageId for a signature, or `None`.
738    pub fn resolve_sig(&self, sig: &str) -> Result<Option<String>, StoreError> {
739        let life = match self.read_lifecycle(sig) {
740            Ok(l) => l,
741            Err(_) => return Ok(None),
742        };
743        Ok(life.current_active().map(|s| s.to_string()))
744    }
745
746    /// Per-stage history for a SigId, ordered chronologically by
747    /// the *last* transition timestamp. Returns one entry per
748    /// distinct StageId that has ever been published under `sig`.
749    /// `Ok(vec![])` if the SigId doesn't exist in the store.
750    ///
751    /// Used by `lex blame` to render "where does this fn come from".
752    pub fn sig_history(&self, sig: &str) -> Result<Vec<StageHistoryEntry>, StoreError> {
753        let life = match self.read_lifecycle(sig) {
754            Ok(l) => l,
755            Err(_) => return Ok(Vec::new()),
756        };
757        // Collapse transitions: latest status + last_at per stage,
758        // plus the timestamp of the first Draft transition (≈ when
759        // the stage was published) when one exists.
760        let mut by_stage: indexmap::IndexMap<String, StageHistoryEntry> = indexmap::IndexMap::new();
761        for t in &life.transitions {
762            let entry = by_stage
763                .entry(t.stage_id.clone())
764                .or_insert(StageHistoryEntry {
765                    stage_id: t.stage_id.clone(),
766                    status: t.to,
767                    last_at: t.at,
768                    published_at: None,
769                });
770            entry.status = t.to;
771            entry.last_at = t.at;
772            if t.from == StageStatus::Draft && entry.published_at.is_none() {
773                entry.published_at = Some(t.at);
774            }
775            if t.to == StageStatus::Draft && entry.published_at.is_none() {
776                // Initial publication: Draft is the *destination*.
777                entry.published_at = Some(t.at);
778            }
779        }
780        let mut out: Vec<StageHistoryEntry> = by_stage.into_values().collect();
781        // Sort newest first so `lex blame` shows recent activity at top.
782        out.sort_by_key(|e| std::cmp::Reverse(e.last_at));
783        Ok(out)
784    }
785
786    pub fn get_ast(&self, stage_id: &str) -> Result<Stage, StoreError> {
787        let (sig, _) = self.lookup_lifecycle(stage_id)?;
788        let bytes = self.read_stage_canonical_bytes(&sig, stage_id)?;
789        Ok(serde_json::from_slice(&bytes)?)
790    }
791
792    /// Bulk AST fetch for callers that already know each stage's
793    /// **signature** — a branch head map, for instance, which is keyed
794    /// by SigId and whose values are the StageIds it points at.
795    ///
796    /// Prefer this over [`Self::get_asts_bulk`] whenever the SigId is in
797    /// hand, because resolving a StageId back to a SigId is not
798    /// reliable: a StageId hashes the structural signature plus the
799    /// implementation, deliberately *not* the name
800    /// (`docs/INVARIANTS.md`), so two functions that differ only in name
801    /// share one StageId while having two distinct SigIds — and two
802    /// separate ASTs, one under each sig directory. `stage_index` maps
803    /// each StageId to a single sig, so `get_ast`/`get_asts_bulk` return
804    /// whichever of those ASTs the index happens to name, i.e. the wrong
805    /// name half the time (#826). Reading straight from the sig the
806    /// caller already knows removes the ambiguity — and skips loading
807    /// the index at all.
808    ///
809    /// Returns results in the same order as `pairs`, `Err` for anything
810    /// that fails to resolve (mirroring `get_ast`'s error semantics).
811    pub fn get_asts_for_sigs_bulk(
812        &self,
813        pairs: &[(String, String)],
814    ) -> Vec<Result<Stage, StoreError>> {
815        pairs
816            .iter()
817            .map(|(sig_id, stage_id)| {
818                let bytes = self.read_stage_canonical_bytes(sig_id, stage_id)?;
819                Ok(serde_json::from_slice(&bytes)?)
820            })
821            .collect()
822    }
823
824    /// Bulk variant of [`Self::get_ast`] for callers resolving many
825    /// stage_ids at once (e.g. `pkg_publish_handler`'s `old_head`
826    /// scan over every live function in a tenant, once per publish
827    /// request). `get_ast` in a loop calls `lookup_lifecycle` once
828    /// per stage_id, and `lookup_lifecycle`'s index-hit path reads
829    /// and re-parses the *entire* `stage_index.jsonl` on every single
830    /// call — fine for one call, but O(index size × N) for N calls in
831    /// a row, which dominates once the index itself is large (#825's
832    /// follow-up: still correct and far better than the pre-index
833    /// full-tenant-scan-per-call behavior, but the per-call reparse
834    /// is itself a real, measured cost — 87.6s for 3,664 calls against
835    /// a ~14k-line index on the alpibrusl tenant).
836    ///
837    /// This loads the index once for the whole batch and keeps it in
838    /// memory across all `stage_ids`, only touching disk again to
839    /// append genuinely new entries (a positive backfill or a
840    /// negative "not found anywhere" cache, same as the single-call
841    /// path) — never to re-read what's already loaded.
842    ///
843    /// Returns results in the same order as `stage_ids`, `Err` for
844    /// anything that fails to resolve (mirroring `get_ast`'s error
845    /// semantics per call).
846    pub fn get_asts_bulk(&self, stage_ids: &[String]) -> Vec<Result<Stage, StoreError>> {
847        let mut index = self.load_stage_index();
848        let mut sigs_cache: BTreeMap<String, Option<Lifecycle>> = BTreeMap::new();
849        let mut all_sigs: Option<Vec<String>> = None;
850
851        stage_ids
852            .iter()
853            .map(|stage_id| {
854                self.lookup_lifecycle_bulk(stage_id, &mut index, &mut sigs_cache, &mut all_sigs)
855                    .and_then(|sig| {
856                        let bytes = self.read_stage_canonical_bytes(&sig, stage_id)?;
857                        Ok(serde_json::from_slice(&bytes)?)
858                    })
859            })
860            .collect()
861    }
862
863    /// Shared implementation behind [`Self::get_asts_bulk`]: identical
864    /// logic to [`Self::lookup_lifecycle`], but reads and writes the
865    /// caller-supplied `index` map instead of reloading it from disk
866    /// on every call, and memoizes `read_lifecycle` per sig and the
867    /// `list_sigs()` full-scan list across the whole batch. Disk
868    /// writes for newly-discovered entries (positive or negative)
869    /// still happen immediately, same as the single-call path — only
870    /// the repeated *reads* are batched away.
871    fn lookup_lifecycle_bulk(
872        &self,
873        stage_id: &str,
874        index: &mut BTreeMap<String, String>,
875        sigs_cache: &mut BTreeMap<String, Option<Lifecycle>>,
876        all_sigs: &mut Option<Vec<String>>,
877    ) -> Result<String, StoreError> {
878        if let Some(sig) = index.get(stage_id) {
879            if sig == MISSING_STAGE_MARKER {
880                return Err(StoreError::UnknownStage(stage_id.into()));
881            }
882            let life = sigs_cache
883                .entry(sig.clone())
884                .or_insert_with(|| self.read_lifecycle(sig).ok());
885            if let Some(life) = life {
886                if life.transitions.iter().any(|t| t.stage_id == stage_id) {
887                    return Ok(sig.clone());
888                }
889            }
890        }
891        if all_sigs.is_none() {
892            *all_sigs = Some(self.list_sigs()?);
893        }
894        for sig in all_sigs.as_ref().unwrap() {
895            let life = sigs_cache
896                .entry(sig.clone())
897                .or_insert_with(|| self.read_lifecycle(sig).ok());
898            if let Some(life) = life {
899                if life.transitions.iter().any(|t| t.stage_id == stage_id) {
900                    self.append_stage_index_entry(stage_id, sig);
901                    index.insert(stage_id.to_string(), sig.clone());
902                    return Ok(sig.clone());
903                }
904            }
905        }
906        self.append_stage_index_entry(stage_id, MISSING_STAGE_MARKER);
907        index.insert(stage_id.to_string(), MISSING_STAGE_MARKER.to_string());
908        Err(StoreError::UnknownStage(stage_id.into()))
909    }
910
911    /// Read the canonical bytes of a stage, walking back through
912    /// any delta chain (#261 slice 3). The recursion ends at a
913    /// `<stage_id>.ast.json` file (a full snapshot) or, in the
914    /// degenerate case of a missing chain, with `UnknownStage`.
915    fn read_stage_canonical_bytes(&self, sig: &str, stage_id: &str) -> Result<Vec<u8>, StoreError> {
916        let ast_path = self.impl_dir(sig).join(format!("{}.ast.json", stage_id));
917        if ast_path.exists() {
918            return Ok(fs::read(&ast_path)?);
919        }
920        let delta_path = self.impl_dir(sig).join(format!("{}.delta.json", stage_id));
921        if !delta_path.exists() {
922            return Err(StoreError::UnknownStage(stage_id.into()));
923        }
924        let delta_bytes = fs::read(&delta_path)?;
925        let delta: crate::delta::StageDelta = serde_json::from_slice(&delta_bytes)?;
926        let base_bytes = self.read_stage_canonical_bytes(sig, &delta.base_stage_id)?;
927        crate::delta::apply(&base_bytes, &delta).map_err(|e| {
928            StoreError::Io(std::io::Error::new(
929                std::io::ErrorKind::InvalidData,
930                format!("applying delta for {stage_id}: {e}"),
931            ))
932        })
933    }
934
935    /// Persist a freshly-published stage's canonical bytes (#261
936    /// slice 3). Tries delta encoding against the most recent
937    /// prior stage in the sig's lifecycle; falls back to a full
938    /// snapshot when no base exists, the diff ratio is too high,
939    /// or the delta chain is already at its cap.
940    fn persist_stage_bytes(
941        &self,
942        sig: &str,
943        stage_id: &str,
944        stage: &Stage,
945        ast_path: &Path,
946        delta_path: &Path,
947    ) -> Result<(), StoreError> {
948        let new_bytes = canonical_bytes(stage)?;
949        if let Some((base_stage_id, base_chain_length)) = self.pick_delta_base(sig, stage_id)? {
950            let base_bytes = self.read_stage_canonical_bytes(sig, &base_stage_id)?;
951            let (prefix, suffix, middle) = crate::delta::splice(&base_bytes, &new_bytes);
952            let chain_length = base_chain_length + 1;
953            if crate::delta::is_worth_encoding(middle.len(), new_bytes.len(), chain_length) {
954                let delta = crate::delta::StageDelta {
955                    base_stage_id,
956                    chain_length,
957                    common_prefix: prefix,
958                    common_suffix: suffix,
959                    middle_hex: hex::encode(&middle),
960                };
961                write_canonical_json(delta_path, &delta)?;
962                return Ok(());
963            }
964        }
965        // Fall through: full snapshot.
966        if let Some(parent) = ast_path.parent() {
967            fs::create_dir_all(parent)?;
968        }
969        fs::write(ast_path, &new_bytes)?;
970        Ok(())
971    }
972
973    /// Pick a base stage for delta encoding from the given sig's
974    /// lifecycle. Returns `(base_stage_id, base_chain_length)` for
975    /// the most-recent non-tombstoned prior stage, or `None` when
976    /// there is no candidate. The chain length is read off the
977    /// base's `.delta.json` (if any) to enforce the cap.
978    fn pick_delta_base(
979        &self,
980        sig: &str,
981        new_stage_id: &str,
982    ) -> Result<Option<(String, usize)>, StoreError> {
983        let life = self.read_lifecycle(sig).ok();
984        let Some(life) = life else {
985            return Ok(None);
986        };
987        // Walk transitions newest-first; pick the first prior
988        // stage that isn't this one and isn't tombstoned.
989        let mut latest_per_stage: indexmap::IndexMap<&str, StageStatus> = indexmap::IndexMap::new();
990        for t in &life.transitions {
991            latest_per_stage.insert(&t.stage_id, t.to);
992        }
993        let mut candidates: Vec<&str> = latest_per_stage
994            .iter()
995            .filter(|(id, status)| **id != new_stage_id && **status != StageStatus::Tombstone)
996            .map(|(id, _)| *id)
997            .collect();
998        // Reverse to get newest-first (transitions are append-only,
999        // so latest_per_stage's iteration order matches insertion
1000        // order, oldest-first).
1001        candidates.reverse();
1002        let Some(&base) = candidates.first() else {
1003            return Ok(None);
1004        };
1005        let base_chain_length = self.delta_chain_length(sig, base)?;
1006        Ok(Some((base.to_string(), base_chain_length)))
1007    }
1008
1009    /// Length of the delta chain ending at `stage_id`. Zero when
1010    /// the stage is a full snapshot (`.ast.json` present); the
1011    /// stored `chain_length` from `.delta.json` otherwise.
1012    fn delta_chain_length(&self, sig: &str, stage_id: &str) -> Result<usize, StoreError> {
1013        let ast_path = self.impl_dir(sig).join(format!("{}.ast.json", stage_id));
1014        if ast_path.exists() {
1015            return Ok(0);
1016        }
1017        let delta_path = self.impl_dir(sig).join(format!("{}.delta.json", stage_id));
1018        if !delta_path.exists() {
1019            return Ok(0);
1020        }
1021        let bytes = fs::read(&delta_path)?;
1022        let delta: crate::delta::StageDelta = serde_json::from_slice(&bytes)?;
1023        Ok(delta.chain_length)
1024    }
1025
1026    pub fn get_metadata(&self, stage_id: &str) -> Result<Metadata, StoreError> {
1027        let (sig, _) = self.lookup_lifecycle(stage_id)?;
1028        let path = self
1029            .impl_dir(&sig)
1030            .join(format!("{}.metadata.json", stage_id));
1031        let bytes = fs::read(&path)?;
1032        Ok(serde_json::from_slice(&bytes)?)
1033    }
1034
1035    pub fn get_status(&self, stage_id: &str) -> Result<StageStatus, StoreError> {
1036        let (_sig, life) = self.lookup_lifecycle(stage_id)?;
1037        life.status_of(stage_id)
1038            .ok_or_else(|| StoreError::UnknownStage(stage_id.into()))
1039    }
1040
1041    pub fn list_stages_by_name(&self, name: &str) -> Result<Vec<String>, StoreError> {
1042        // Walk every SigId → check metadata of any implementation; if its
1043        // name matches, include the SigId.
1044        let mut out = Vec::new();
1045        let stages_dir = self.root.join("stages");
1046        if !stages_dir.exists() {
1047            return Ok(out);
1048        }
1049        for entry in fs::read_dir(&stages_dir)? {
1050            let entry = entry?;
1051            let sig_dir = entry.path();
1052            if !sig_dir.is_dir() {
1053                continue;
1054            }
1055            let sig = entry.file_name().to_string_lossy().to_string();
1056            // Look at any one metadata file under this SigId.
1057            let impls = self.impl_dir(&sig);
1058            if !impls.exists() {
1059                continue;
1060            }
1061            for f in fs::read_dir(impls)? {
1062                let f = f?;
1063                let p = f.path();
1064                if p.extension().is_some_and(|e| e == "json")
1065                    && p.file_name()
1066                        .is_some_and(|n| n.to_string_lossy().ends_with(".metadata.json"))
1067                {
1068                    if let Ok(bytes) = fs::read(&p) {
1069                        if let Ok(m) = serde_json::from_slice::<Metadata>(&bytes) {
1070                            if m.name == name {
1071                                if !out.contains(&sig) {
1072                                    out.push(sig.clone());
1073                                }
1074                                break;
1075                            }
1076                        }
1077                    }
1078                }
1079            }
1080        }
1081        out.sort();
1082        Ok(out)
1083    }
1084
1085    pub fn list_sigs(&self) -> Result<Vec<String>, StoreError> {
1086        let stages_dir = self.root.join("stages");
1087        let mut out = Vec::new();
1088        if !stages_dir.exists() {
1089            return Ok(out);
1090        }
1091        for entry in fs::read_dir(stages_dir)? {
1092            let entry = entry?;
1093            if entry.file_type()?.is_dir() {
1094                out.push(entry.file_name().to_string_lossy().to_string());
1095            }
1096        }
1097        out.sort();
1098        Ok(out)
1099    }
1100
1101    // ---- tests/specs as metadata (§4.4) ----
1102
1103    pub fn attach_test(&self, sig: &str, test: &Test) -> Result<String, StoreError> {
1104        if !self.sig_dir(sig).exists() {
1105            return Err(StoreError::UnknownSig(sig.into()));
1106        }
1107        fs::create_dir_all(self.tests_dir(sig))?;
1108        let path = self.tests_dir(sig).join(format!("{}.json", test.id));
1109        write_canonical_json(&path, test)?;
1110        Ok(test.id.clone())
1111    }
1112
1113    pub fn list_tests(&self, sig: &str) -> Result<Vec<Test>, StoreError> {
1114        let dir = self.tests_dir(sig);
1115        if !dir.exists() {
1116            return Ok(Vec::new());
1117        }
1118        let mut out = Vec::new();
1119        for f in fs::read_dir(dir)? {
1120            let f = f?;
1121            if f.path().extension().is_some_and(|e| e == "json") {
1122                let bytes = fs::read(f.path())?;
1123                out.push(serde_json::from_slice(&bytes)?);
1124            }
1125        }
1126        Ok(out)
1127    }
1128
1129    pub fn attach_spec(&self, sig: &str, spec: &Spec) -> Result<String, StoreError> {
1130        if !self.sig_dir(sig).exists() {
1131            return Err(StoreError::UnknownSig(sig.into()));
1132        }
1133        fs::create_dir_all(self.specs_dir(sig))?;
1134        let path = self.specs_dir(sig).join(format!("{}.json", spec.id));
1135        write_canonical_json(&path, spec)?;
1136        Ok(spec.id.clone())
1137    }
1138
1139    pub fn list_specs(&self, sig: &str) -> Result<Vec<Spec>, StoreError> {
1140        let dir = self.specs_dir(sig);
1141        if !dir.exists() {
1142            return Ok(Vec::new());
1143        }
1144        let mut out = Vec::new();
1145        for f in fs::read_dir(dir)? {
1146            let f = f?;
1147            if f.path().extension().is_some_and(|e| e == "json") {
1148                let bytes = fs::read(f.path())?;
1149                out.push(serde_json::from_slice(&bytes)?);
1150            }
1151        }
1152        Ok(out)
1153    }
1154
1155    // ---- traces (§4.2 / M7) ----
1156
1157    // Native run-trace store — gated behind the `trace` feature (depends on
1158    // lex-trace). Off when a lower crate (lex-runtime) depends on lex-store to
1159    // avoid a dependency cycle; the blob/stage store below is unaffected.
1160    #[cfg(feature = "trace")]
1161    fn trace_path(&self, run_id: &str) -> PathBuf {
1162        self.root.join("traces").join(run_id).join("trace.json")
1163    }
1164
1165    #[cfg(feature = "trace")]
1166    pub fn save_trace(&self, tree: &lex_trace::TraceTree) -> Result<String, StoreError> {
1167        let path = self.trace_path(&tree.run_id);
1168        write_canonical_json(&path, tree)?;
1169        Ok(tree.run_id.clone())
1170    }
1171
1172    #[cfg(feature = "trace")]
1173    pub fn load_trace(&self, run_id: &str) -> Result<lex_trace::TraceTree, StoreError> {
1174        let bytes = fs::read(self.trace_path(run_id))?;
1175        Ok(serde_json::from_slice(&bytes)?)
1176    }
1177
1178    pub fn list_traces(&self) -> Result<Vec<String>, StoreError> {
1179        let dir = self.root.join("traces");
1180        if !dir.exists() {
1181            return Ok(Vec::new());
1182        }
1183        let mut out = Vec::new();
1184        for entry in fs::read_dir(dir)? {
1185            let entry = entry?;
1186            if entry.file_type()?.is_dir() {
1187                out.push(entry.file_name().to_string_lossy().to_string());
1188            }
1189        }
1190        out.sort();
1191        Ok(out)
1192    }
1193
1194    // ---- internals ----
1195
1196    /// `<root>/stage_index.jsonl` — an append-only, best-effort
1197    /// reverse index (`StageId` -> owning `SigId`), one JSON object
1198    /// per line. Backs `lookup_lifecycle`'s fast path; see its doc
1199    /// comment. Not a second source of truth: every entry is
1200    /// reconstructible from `stages/<sig>/lifecycle.json`, so a
1201    /// missing, truncated, or entirely absent index file only costs
1202    /// a slower lookup (the pre-existing full scan), never
1203    /// correctness — matching this module's "filesystem is the
1204    /// source of truth" stance (see the module doc comment) rather
1205    /// than introducing an actual second database.
1206    fn stage_index_path(&self) -> PathBuf {
1207        self.root.join("stage_index.jsonl")
1208    }
1209
1210    /// Best-effort load of the whole reverse index into memory.
1211    /// Tolerates a missing file (no index yet) and a corrupt or
1212    /// torn last line (a crash mid-append under the single-writer
1213    /// Tier-1 assumption) by skipping lines that don't parse,
1214    /// rather than failing the lookup that triggered the load.
1215    fn load_stage_index(&self) -> std::collections::BTreeMap<String, String> {
1216        let mut out = std::collections::BTreeMap::new();
1217        let Ok(raw) = fs::read_to_string(self.stage_index_path()) else {
1218            return out;
1219        };
1220        for line in raw.lines() {
1221            if let Ok(entry) = serde_json::from_str::<StageIndexEntry>(line) {
1222                out.insert(entry.stage_id, entry.sig_id);
1223            }
1224        }
1225        out
1226    }
1227
1228    /// Best-effort append of one new `(stage_id, sig_id)` pair.
1229    /// Failure (e.g. a read-only filesystem) only costs a future
1230    /// full scan for this stage_id, never correctness, so it's
1231    /// swallowed rather than propagated.
1232    fn append_stage_index_entry(&self, stage_id: &str, sig: &str) {
1233        use std::io::Write;
1234        let entry = StageIndexEntry { stage_id: stage_id.into(), sig_id: sig.into() };
1235        let Ok(line) = serde_json::to_string(&entry) else { return };
1236        if let Ok(mut f) = fs::OpenOptions::new()
1237            .create(true)
1238            .append(true)
1239            .open(self.stage_index_path())
1240        {
1241            let _ = writeln!(f, "{line}");
1242        }
1243    }
1244
1245    /// Find which SigId owns a StageId, and that sig's lifecycle.
1246    ///
1247    /// Before the reverse index (#822): a full scan over *every*
1248    /// SigId in the tenant (`list_sigs()`, not scoped to the
1249    /// package being looked at), reading and parsing each one's
1250    /// `lifecycle.json` until a match turned up. `get_ast` — called
1251    /// once per pre-existing function when building a publish
1252    /// request's `old_fns_by_name` (`lex-api/src/handlers.rs`) —
1253    /// calls this once per function, so a tenant with a few thousand
1254    /// published functions turned a single publish into millions of
1255    /// individual file reads; measured at roughly an hour on the
1256    /// `alpibrusl` tenant's ~2,400-function store.
1257    ///
1258    /// Now: check the persisted reverse index first (one sequential
1259    /// file read instead of up to N separate ones). A miss — the
1260    /// index doesn't exist yet, or this stage_id predates it — falls
1261    /// back to the full scan and backfills the index so the next
1262    /// lookup for the same stage_id is fast.
1263    fn lookup_lifecycle(&self, stage_id: &str) -> Result<(String, Lifecycle), StoreError> {
1264        let index = self.load_stage_index();
1265        if let Some(sig) = index.get(stage_id) {
1266            if sig == MISSING_STAGE_MARKER {
1267                // A previous full scan already established this
1268                // stage_id exists nowhere in the store. Re-scanning
1269                // would find nothing again -- see #825: a genuinely
1270                // orphaned reference (e.g. from data predating some
1271                // store migration) is looked up on *every* call that
1272                // needs it, forever, so without this negative cache
1273                // it silently costs a full O(total sigs) scan each
1274                // time, indistinguishable from the positive case at
1275                // the call site. Measured directly: on the alpibrusl
1276                // tenant, 988 of 3,664 branch-head entries are
1277                // orphaned this way, turning one `pkg publish`'s
1278                // old_fns_by_name build into ~16M wasted lifecycle
1279                // reads.
1280                return Err(StoreError::UnknownStage(stage_id.into()));
1281            }
1282            if let Ok(life) = self.read_lifecycle(sig) {
1283                if life.transitions.iter().any(|t| t.stage_id == stage_id) {
1284                    return Ok((sig.clone(), life));
1285                }
1286            }
1287            // Index entry is stale or wrong (shouldn't happen in
1288            // practice — sig ownership of a stage_id is permanent).
1289            // Fall through to the full scan below rather than trust it.
1290        }
1291        for sig in self.list_sigs()? {
1292            if let Ok(life) = self.read_lifecycle(&sig) {
1293                if life.transitions.iter().any(|t| t.stage_id == stage_id) {
1294                    self.append_stage_index_entry(stage_id, &sig);
1295                    return Ok((sig, life));
1296                }
1297            }
1298        }
1299        // Genuinely not found anywhere: cache that fact so the next
1300        // lookup for this exact stage_id is an index hit, not another
1301        // full scan. Safe even if this stage_id somehow gets a real
1302        // sig later (content-addressed publish is idempotent, so
1303        // "later" only means "a byte-identical stage republished
1304        // under a real sig") — `append_stage_index_entry`'s later,
1305        // real entry is a later line in the file, and `load_stage_index`
1306        // folds duplicate keys last-write-wins, so the real entry wins.
1307        self.append_stage_index_entry(stage_id, MISSING_STAGE_MARKER);
1308        Err(StoreError::UnknownStage(stage_id.into()))
1309    }
1310
1311    fn read_lifecycle(&self, sig: &str) -> Result<Lifecycle, StoreError> {
1312        let path = self.lifecycle_path(sig);
1313        if !path.exists() {
1314            return Ok(Lifecycle {
1315                sig_id: sig.into(),
1316                transitions: Vec::new(),
1317            });
1318        }
1319        let bytes = fs::read(&path)?;
1320        Ok(serde_json::from_slice(&bytes)?)
1321    }
1322
1323    fn write_lifecycle(&self, sig: &str, life: &Lifecycle) -> Result<(), StoreError> {
1324        write_canonical_json(&self.lifecycle_path(sig), life)
1325    }
1326
1327    /// Apply a published program to a branch as a sequence of typed
1328    /// operations. Returns the ordered list of op_ids + the new
1329    /// head_op. The caller (`lex publish` CLI, `lex serve`'s HTTP
1330    /// handler) is responsible for computing the `DiffReport` against
1331    /// the current branch head — the diff infrastructure lives in
1332    /// `lex-vcs::compute_diff` (previously `lex-cli`) to keep this
1333    /// layer from owning diffing logic.
1334    ///
1335    /// On success: every op in the returned list is durable in the
1336    /// op log and the branch's head_op points at the last one.
1337    /// On a no-op (no diff): returns empty `ops` and the existing
1338    /// `head_op` unchanged.
1339    pub fn publish_program(
1340        &self,
1341        branch: &str,
1342        stages: &[lex_ast::Stage],
1343        diff: &lex_vcs::DiffReport,
1344        new_imports: &lex_vcs::ImportMap,
1345        activate: bool,
1346    ) -> Result<PublishOutcome, StoreError> {
1347        self.publish_program_signed(branch, stages, diff, new_imports, activate, None)
1348    }
1349
1350    /// Signed variant of [`Self::publish_program`] (#227). Every
1351    /// stage written under this batch gets the same signer; per-stage
1352    /// keys aren't supported because the agent identity model treats
1353    /// a publish as a single authorial act.
1354    pub fn publish_program_signed(
1355        &self,
1356        branch: &str,
1357        stages: &[lex_ast::Stage],
1358        diff: &lex_vcs::DiffReport,
1359        new_imports: &lex_vcs::ImportMap,
1360        activate: bool,
1361        signer: Option<&lex_vcs::Keypair>,
1362    ) -> Result<PublishOutcome, StoreError> {
1363        // Single-file / test callers don't publish a mangled package, so
1364        // there are no module prefixes to record (`in_file` stays `None`).
1365        self.publish_program_with_intent(
1366            branch,
1367            stages,
1368            diff,
1369            new_imports,
1370            activate,
1371            signer,
1372            None,
1373            &std::collections::BTreeMap::new(),
1374        )
1375    }
1376
1377    /// [`Self::publish_program_signed`] plus an optional `intent_id`
1378    /// (#131 / #839): when given, every op this publish emits is stamped
1379    /// with it, so the op log records *why* the change happened — the
1380    /// prompt / model / session an agent was acting under — not only
1381    /// what it was. `lex recall --intent <id>` and `lex op replay` read
1382    /// it back. The caller records the [`lex_vcs::Intent`] in the
1383    /// [`lex_vcs::IntentLog`] beforehand; this only links ops to it.
1384    /// `None` is the existing (intent-less) behavior, so op ids for
1385    /// intent-less publishes are unchanged.
1386    // A batch publish legitimately takes the branch, program, diff,
1387    // imports, activate flag, signer, and now the intent — bundling
1388    // them into a struct for one optional field would obscure more
1389    // than it clarifies.
1390    #[allow(clippy::too_many_arguments)]
1391    pub fn publish_program_with_intent(
1392        &self,
1393        branch: &str,
1394        stages: &[lex_ast::Stage],
1395        diff: &lex_vcs::DiffReport,
1396        new_imports: &lex_vcs::ImportMap,
1397        activate: bool,
1398        signer: Option<&lex_vcs::Keypair>,
1399        intent_id: Option<lex_vcs::IntentId>,
1400        // Mangling prefix → package source file, for a multi-module
1401        // package publish; empty for a single file. Recorded as each
1402        // `AddFunction`/`AddType`'s `in_file` so `export-git` can
1403        // de-flatten the package (#894).
1404        module_prefixes: &std::collections::BTreeMap<String, String>,
1405    ) -> Result<PublishOutcome, StoreError> {
1406        use std::collections::{BTreeMap, BTreeSet};
1407
1408        // #130's write-time gate: verify the candidate program
1409        // typechecks (and effects are correctly declared) before
1410        // any disk side-effect. If anything fails, return the
1411        // structured envelope and leave the branch head unchanged
1412        // — the store's "always-valid HEAD" invariant only holds
1413        // because this is the only batch-publish path that
1414        // advances heads. Single-op writes via the lower-level
1415        // `apply_operation` are not gated yet (#130 follow-up).
1416        // #930: resolve any external dependency edges the head keeps
1417        // (empty when no resolver is installed or the head is inlined).
1418        let modules = self.resolved_modules(stages, None);
1419        if let Err(errors) = lex_types::check_program_with_modules(stages, &modules) {
1420            return Err(StoreError::TypeError(errors));
1421        }
1422
1423        // Build old-side views from the current branch. There used to be
1424        // an `old_name_to_sig: BTreeMap<String, SigId>` built here too,
1425        // keyed by bare function name — but a bare name is not unique
1426        // across a package's files (#818: two files can legitimately
1427        // both declare a local `validate` helper with different
1428        // signatures), so a name-keyed map silently collapsed distinct
1429        // SigIds onto one. `diff` now carries each entry's own resolved
1430        // `old_sig_id` directly (see `diff_report`'s doc comments), so
1431        // `diff_to_ops` no longer needs this lookup at all.
1432        let old_head = self.branch_head(branch)?;
1433        // Read every live function's effects through the SigId the head
1434        // names, in one batch. Two reasons, both load-bearing:
1435        //
1436        //   * Cost. This was a `get_ast` per live function, and
1437        //     `get_ast`'s index-hit path re-reads and re-parses the whole
1438        //     `stage_index.jsonl` on every call — O(index × live fns) per
1439        //     publish, paid again for every `publish_program` call a
1440        //     multi-file publish makes (#828; measured 34s for a no-op
1441        //     republish of a real 21-file package against only 698 live
1442        //     functions, nearly all of it here).
1443        //   * Correctness. A StageId is name-independent, so two live
1444        //     functions differing only in name share one and the index
1445        //     maps it to a single sig — resolving by StageId therefore
1446        //     attributed one function's effects to the *other* one's sig,
1447        //     the same ambiguity #826 fixed in `pkg_publish_handler`.
1448        let head_pairs: Vec<(String, String)> = old_head
1449            .iter()
1450            .map(|(sig, stage)| (sig.clone(), stage.clone()))
1451            .collect();
1452        let old_effects: BTreeMap<String, BTreeSet<String>> = head_pairs
1453            .iter()
1454            .zip(self.get_asts_for_sigs_bulk(&head_pairs))
1455            .filter_map(|((sig, _), ast)| match ast.ok()? {
1456                lex_ast::Stage::FnDecl(fd) => {
1457                    let s: BTreeSet<String> =
1458                        fd.effects.iter().map(|e| e.name.clone()).collect();
1459                    Some((sig.clone(), s))
1460                }
1461                _ => None,
1462            })
1463            .collect();
1464        let old_imports = self.derive_imports_from_oplog(branch)?;
1465
1466        let op_kinds = lex_vcs::diff_to_ops(lex_vcs::DiffInputs {
1467            old_head: &old_head,
1468            old_effects: &old_effects,
1469            old_imports: &old_imports,
1470            new_stages: stages,
1471            new_imports,
1472            diff,
1473            module_prefixes,
1474        })
1475        .map_err(|e| StoreError::InvalidTransition(format!("diff_to_ops: {e}")))?;
1476
1477        let mut ops_out: Vec<PublishOp> = Vec::new();
1478        let mut last_op_id: Option<lex_vcs::OpId> = None;
1479        for kind in op_kinds {
1480            // Persist the underlying stage AST/metadata if this op
1481            // produces or replaces one.
1482            if let Some(stg) = stage_for_kind(&kind, stages) {
1483                if !matches!(stg, lex_ast::Stage::Import(_)) {
1484                    self.publish_signed(stg, signer)?;
1485                    if activate {
1486                        if let Some(stage_id_str) = stage_id(stg) {
1487                            let _ = self.activate(&stage_id_str);
1488                        }
1489                    }
1490                }
1491            }
1492            let transition = transition_for_kind(&kind);
1493            let attestable = attestable_stage_ids(&transition);
1494            let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
1495            let op =
1496                lex_vcs::Operation::new(kind.clone(), head_now.into_iter().collect::<Vec<_>>());
1497            // #131 / #839: stamp the caller's intent so the op log records
1498            // why this change happened, not just what it was. The CAS
1499            // retry path preserves `intent_id` when it rebuilds the op.
1500            let op = match &intent_id {
1501                Some(id) => op.with_intent(id.clone()),
1502                None => op,
1503            };
1504            let op_id = self.apply_operation(branch, op, transition)?;
1505            self.record_typecheck_passed(&attestable, &op_id)?;
1506            ops_out.push(PublishOp {
1507                op_id: op_id.clone(),
1508                kind: serde_json::to_value(&kind).map_err(StoreError::Serde)?,
1509            });
1510            last_op_id = Some(op_id);
1511        }
1512
1513        let head_op = match last_op_id {
1514            Some(id) => Some(id),
1515            // No ops applied; return whatever the head was already.
1516            None => self.get_branch(branch)?.and_then(|b| b.head_op),
1517        };
1518
1519        Ok(PublishOutcome {
1520            ops: ops_out,
1521            head_op,
1522        })
1523    }
1524
1525    pub fn derive_imports_from_oplog(
1526        &self,
1527        branch: &str,
1528    ) -> Result<lex_vcs::ImportMap, StoreError> {
1529        use lex_vcs::OperationKind::*;
1530        let log = lex_vcs::OpLog::open(self.root())?;
1531        let head = match self.get_branch(branch)?.and_then(|b| b.head_op) {
1532            Some(h) => h,
1533            None => return Ok(Default::default()),
1534        };
1535        let mut out: lex_vcs::ImportMap = Default::default();
1536        for r in log.walk_forward(&head, None)? {
1537            match r.op.kind {
1538                AddImport { in_file, module, alias } => {
1539                    // The op omits the alias when it's the module's
1540                    // default (last path segment) to keep its OpId
1541                    // stable; rebuild it the same way on the way out.
1542                    let alias =
1543                        alias.unwrap_or_else(|| lex_vcs::default_import_alias(&module));
1544                    out.entry(in_file)
1545                        .or_default()
1546                        .insert(lex_vcs::ImportRef { reference: module, alias });
1547                }
1548                RemoveImport { in_file, module } => {
1549                    // Removal is keyed by reference (the op carries no
1550                    // alias), so drop any binding of that module.
1551                    if let Some(set) = out.get_mut(&in_file) {
1552                        set.retain(|ir| ir.reference != module);
1553                    }
1554                }
1555                _ => {}
1556            }
1557        }
1558        Ok(out)
1559    }
1560
1561    /// Apply an operation to a branch and advance its head_op.
1562    ///
1563    /// The single advance path. Validates parents via `lex_vcs::apply`,
1564    /// persists the operation via the op log, then atomically advances
1565    /// the branch file's head_op via `set_branch_head_op`.
1566    ///
1567    /// Errors:
1568    /// - `UnknownBranch`: branch does not exist (no op is persisted).
1569    /// - `Apply(ApplyError::StaleParent)`: the op's parents don't
1570    ///   match the branch head — head is unchanged. Callers that
1571    ///   want retry-on-stale (e.g. `lex publish` re-running against
1572    ///   a moved head) match on this variant explicitly.
1573    /// - `Apply(ApplyError::UnknownMergeParent)`: a merge op's
1574    ///   second parent isn't in the log.
1575    /// - `Io`: filesystem error during persist or branch advance.
1576    ///
1577    /// Crash recovery: between op persist and branch advance, a crash
1578    /// can leave an orphan op record in the log with no branch
1579    /// pointing at it. The op is content-addressed and cheap to
1580    /// re-derive from the same source. See
1581    /// Apply a single op against `branch`, gated on the candidate
1582    /// program typechecking. The per-op variant of #130's
1583    /// write-time gate — counterpart to [`Self::publish_program`]'s
1584    /// batch-mode check.
1585    ///
1586    /// `candidate` is the sequence of `Stage`s that *would* exist
1587    /// on this branch after the op is applied. Caller's
1588    /// responsibility: today neither `lex-store` nor `lex-vcs`
1589    /// reconstruct the candidate from the op + branch state on
1590    /// behalf of the caller. The natural callers (HTTP `POST
1591    /// /v1/publish` for a single op; agent harnesses driving
1592    /// merges via the future #134 API) already have the candidate
1593    /// in memory.
1594    ///
1595    /// On rejection: branch head unchanged, no op record persisted.
1596    /// Same atomicity guarantee as the publish path.
1597    ///
1598    /// # Why a separate method, not a flag on `apply_operation`
1599    ///
1600    /// `apply_operation` accepting `Option<&[Stage]>` and silently
1601    /// skipping the gate on `None` is exactly the kind of
1602    /// "secretly opt-out" path #130 is trying to remove. The honest
1603    /// split: `apply_operation` for the one caller that already
1604    /// typechecked its input up front (`publish_program`),
1605    /// `apply_operation_checked` for callers holding the candidate,
1606    /// [`Self::apply_operation_gated`] for single-parent callers
1607    /// that hold only the transition (`/v1/patch`), and
1608    /// [`Self::apply_merge_op_gated`] for merge commits (#833).
1609    pub fn apply_operation_checked(
1610        &self,
1611        branch: &str,
1612        op: lex_vcs::Operation,
1613        transition: lex_vcs::StageTransition,
1614        candidate: &[lex_ast::Stage],
1615    ) -> Result<lex_vcs::OpId, StoreError> {
1616        let modules = self.resolved_modules(candidate, None); // #930
1617        if let Err(errors) = lex_types::check_program_with_modules(candidate, &modules) {
1618            // #281: emit a `RepairHint` attestation against each
1619            // candidate stage the transition was about to produce.
1620            // The op record itself isn't persisted (the gate is
1621            // pre-persistence), but the candidate stage IS — the
1622            // transform-flow methods publish before this call.
1623            // The attached hint lets `lex repair <op_id>` and
1624            // future LLM-assisted apply paths read the structured
1625            // errors without re-running the typecheck.
1626            let attestable = attestable_stage_ids(&transition);
1627            let failed_op_id = op.op_id();
1628            let _ = self.record_repair_hint(&attestable, &failed_op_id, &errors);
1629            return Err(StoreError::TypeError(errors));
1630        }
1631        // #292 slice 3: per-session budget gate. After typecheck
1632        // passes, refuse the op if it would push its session's
1633        // monotonic spend over the configured cap. Sessions
1634        // without an intent_id, or with an intent whose session
1635        // has no cap configured, sail through.
1636        self.check_session_budget(&op)?;
1637        let attestable = attestable_stage_ids(&transition);
1638        let op_effects = op_declared_effects(&op.kind);
1639        // #262: CAS retry loop. Single-parent ops can be safely
1640        // re-persisted under a new parent on contention (the kind
1641        // is invariant; only `parents` changes). Merge ops (already
1642        // 2-parent) come through the merge engine which has its own
1643        // coordination; we don't retry them here — we'll see the
1644        // first attempt's CAS fail and surface Contention.
1645        self.cas_retry_advance(branch, op, transition, |new_head| {
1646            self.record_typecheck_passed(&attestable, &new_head.op_id)?;
1647            self.run_required_attestations_gate(branch, &new_head.op_id, &attestable, &op_effects)
1648        })
1649    }
1650
1651    /// The program that would exist on `branch` after `transition`
1652    /// is applied: the branch head (snapshot-cached) with the
1653    /// transition replayed over it, every resulting `(sig, stage)`
1654    /// bulk-loaded. Exact for a **single-parent** transition — the
1655    /// candidate [`Self::apply_operation_gated`] wants. Not valid for
1656    /// a merge: a `StageTransition::Merge` records only the delta
1657    /// relative to dst, while the op-DAG replay that computes a
1658    /// merge's real head walks both parents (#833).
1659    pub fn candidate_program_for(
1660        &self,
1661        branch: &str,
1662        transition: &lex_vcs::StageTransition,
1663    ) -> Result<Vec<Stage>, StoreError> {
1664        let mut head = self.branch_head(branch)?;
1665        crate::branches::apply_transition(&mut head, transition);
1666        let pairs: Vec<(String, String)> = head.into_iter().collect();
1667        self.get_asts_for_sigs_bulk(&pairs).into_iter().collect()
1668    }
1669
1670    /// [`Self::apply_operation_checked`] for a **single-parent** op
1671    /// where the caller holds only the transition: assembles the
1672    /// candidate via [`Self::candidate_program_for`] and runs the
1673    /// gate. Same rejection semantics — `TypeError`, a `RepairHint`
1674    /// attestation, head unchanged, nothing persisted. This is the
1675    /// write path for `/v1/patch` (#833). Merge ops must not use it
1676    /// (see `candidate_program_for`); they go through
1677    /// [`Self::apply_merge_op_gated`].
1678    pub fn apply_operation_gated(
1679        &self,
1680        branch: &str,
1681        op: lex_vcs::Operation,
1682        transition: lex_vcs::StageTransition,
1683    ) -> Result<lex_vcs::OpId, StoreError> {
1684        debug_assert!(
1685            op.parents.len() <= 1,
1686            "apply_operation_gated is single-parent only; merges use apply_merge_op_gated"
1687        );
1688        let candidate = self.candidate_program_for(branch, &transition)?;
1689        self.apply_operation_checked(branch, op, transition, &candidate)
1690    }
1691
1692    /// The gated write path for **merge** commits (`commit_merge`,
1693    /// `POST /v1/merge/<id>/commit`, `lex merge commit`).
1694    ///
1695    /// A `StageTransition::Merge` records only the delta relative to
1696    /// dst; the sig->stage map every consumer reads is recomputed by
1697    /// replaying the op DAG, which for a merge walks *both* parents
1698    /// and can surface sigs the delta never mentions. So the only way
1699    /// to know the true post-merge program is to replay it — land the
1700    /// op and read `branch_head`. This lands the merge op,
1701    /// type-checks the resulting head, and on a failure rolls the
1702    /// head back and returns `TypeError`.
1703    ///
1704    /// Before #833 the merge paths landed through the ungated
1705    /// `apply_operation`, so a merge whose result didn't compose
1706    /// (e.g. dst still calls `helper`, an agent-supplied resolution
1707    /// dropped it) advanced the head with nothing to catch it.
1708    ///
1709    /// Rollback leaves the rejected merge op as an unreachable record
1710    /// (reclaimed by `lex op gc`, the same orphan crash-recovery
1711    /// already tolerates). A stage the merge names that was never
1712    /// published surfaces as the underlying `StoreError` from the
1713    /// bulk read — the "never advance onto content that can't be
1714    /// loaded" invariant from the other side.
1715    pub fn apply_merge_op_gated(
1716        &self,
1717        branch: &str,
1718        op: lex_vcs::Operation,
1719        transition: lex_vcs::StageTransition,
1720    ) -> Result<lex_vcs::OpId, StoreError> {
1721        let head_before = self.get_branch(branch)?.and_then(|b| b.head_op);
1722        // Capture the stages this merge introduces before `transition`
1723        // is moved into `apply_operation`; used for the TypeCheck
1724        // attestation below.
1725        let attestable = attestable_stage_ids(&transition);
1726        let op_id = self.apply_operation(branch, op, transition)?;
1727
1728        let verdict = (|| -> Result<(), StoreError> {
1729            let head = self.branch_head(branch)?;
1730            let pairs: Vec<(String, String)> = head.into_iter().collect();
1731            let stages: Vec<Stage> =
1732                self.get_asts_for_sigs_bulk(&pairs).into_iter().collect::<Result<_, _>>()?;
1733            // #930: per-head dep resolution for the merge path is a follow-up;
1734            // None lets a client resolver use its working-copy lock.
1735            let modules = self.resolved_modules(&stages, None);
1736            if let Err(errors) = lex_types::check_program_with_modules(&stages, &modules) {
1737                return Err(StoreError::TypeError(errors));
1738            }
1739            Ok(())
1740        })();
1741
1742        if let Err(e) = verdict {
1743            // Roll the head back. The empty-dst case never reaches
1744            // here (it fast-forwards without a merge op), so
1745            // `head_before` is always `Some` on this arm.
1746            if let Some(prev) = head_before {
1747                self.set_branch_head_op(branch, prev)?;
1748            }
1749            return Err(e);
1750        }
1751        // #835: the merge's post-merge head type-checked, but until now
1752        // that verdict left no trace in the attestation log — so a
1753        // merged stage looked un-type-checked to `lex blame
1754        // --with-evidence` and the attestation queries, unlike a
1755        // published or patched stage. Emit `TypeCheck::Passed` for the
1756        // stages the merge introduced, mirroring the publish / patch
1757        // paths (`record_typecheck_passed`). Emitted only after the
1758        // check passes and the head is committed, so a rolled-back
1759        // merge records nothing.
1760        self.record_typecheck_passed(&attestable, &op_id)?;
1761        Ok(op_id)
1762    }
1763
1764    /// Type-check the program that would result from overlaying a merge
1765    /// `delta` onto `branch`'s current head — **without moving the
1766    /// head** (#834). `delta` maps `sig_id -> Some(stage)` to set that
1767    /// sig to `stage`, or `sig_id -> None` to remove it, exactly the
1768    /// `entries` a `StageTransition::Merge` records.
1769    ///
1770    /// This is the read-only, resolve-time counterpart of
1771    /// `apply_merge_op_gated`'s commit-time gate: it lets a merge
1772    /// session tell an agent *which resolution broke type-checking* the
1773    /// moment it is submitted, instead of only after a failed commit.
1774    /// `Ok(())` means the projected program composes; a type failure is
1775    /// `Err(StoreError::TypeError(..))`; a read failure is the
1776    /// corresponding `StoreError` I/O variant.
1777    pub fn typecheck_merge_projection(
1778        &self,
1779        branch: &str,
1780        delta: &std::collections::BTreeMap<String, Option<String>>,
1781    ) -> Result<(), StoreError> {
1782        let mut head = self.branch_head(branch)?;
1783        for (sig, stage) in delta {
1784            match stage {
1785                Some(s) => { head.insert(sig.clone(), s.clone()); }
1786                None => { head.remove(sig); }
1787            }
1788        }
1789        let pairs: Vec<(String, String)> = head.into_iter().collect();
1790        let stages: Vec<Stage> =
1791            self.get_asts_for_sigs_bulk(&pairs).into_iter().collect::<Result<_, _>>()?;
1792        let modules = self.resolved_modules(&stages, None); // #930 (patch path)
1793        if let Err(errors) = lex_types::check_program_with_modules(&stages, &modules) {
1794            return Err(StoreError::TypeError(errors));
1795        }
1796        Ok(())
1797    }
1798
1799    /// #838: attempt a typed three-way merge of a single sig's body for
1800    /// a `ModifyModify` conflict — the intra-function, better-than-git
1801    /// case where two agents edited *disjoint* subtrees of the same
1802    /// function (different match arms, different let bindings).
1803    ///
1804    /// `base` / `ours` (the dst side) / `theirs` (the src side) are the
1805    /// three stage ids the merge engine surfaced for `sig_id`. Loads
1806    /// the three `FnDecl`s, structurally merges the bodies
1807    /// ([`lex_vcs::merge_bodies`]), and accepts the result *only if* the
1808    /// merged function also type-checks against `dst_branch`'s head — a
1809    /// body that composes syntactically but not by type is still a
1810    /// conflict (#838). On success the merged stage is published
1811    /// (content-addressed, idempotent; orphaned and GC-reclaimable if
1812    /// the merge is never committed) and its id returned; `None` means
1813    /// "fall back to a whole-function conflict."
1814    ///
1815    /// Deliberately narrow for this slice: only pure body divergence is
1816    /// merged. If the two sides disagree on anything but the body
1817    /// (examples, type params — the signature is identical by
1818    /// construction, since all three share `sig_id`), or either stage
1819    /// isn't a function, it falls back to a conflict.
1820    pub fn try_semantic_body_merge(
1821        &self,
1822        dst_branch: &str,
1823        sig_id: &str,
1824        base: &str,
1825        ours: &str,
1826        theirs: &str,
1827    ) -> Result<Option<String>, StoreError> {
1828        use lex_ast::Stage::FnDecl;
1829        let (base_fd, ours_fd, theirs_fd) =
1830            match (self.get_ast(base), self.get_ast(ours), self.get_ast(theirs)) {
1831                (Ok(FnDecl(b)), Ok(FnDecl(o)), Ok(FnDecl(t))) => (b, o, t),
1832                // A non-function stage (type decl / import) or a stage
1833                // that can't be loaded isn't an intra-body merge.
1834                _ => return Ok(None),
1835            };
1836
1837        // Only the body may diverge between the two sides.
1838        if !fndecl_same_except_body(&ours_fd, &theirs_fd) {
1839            return Ok(None);
1840        }
1841
1842        let merged_body =
1843            match lex_vcs::merge_bodies(&base_fd.body, &ours_fd.body, &theirs_fd.body) {
1844                lex_vcs::BodyMerge::Merged(b) => b,
1845                lex_vcs::BodyMerge::Conflict => return Ok(None),
1846            };
1847
1848        let mut merged_fd = ours_fd.clone();
1849        merged_fd.body = merged_body;
1850        let merged_stage = lex_ast::Stage::FnDecl(merged_fd);
1851        let new_stage_id = match stage_id(&merged_stage) {
1852            Some(id) => id,
1853            None => return Ok(None),
1854        };
1855
1856        // Type-check the merged fn in context: dst's head with this sig
1857        // swapped to the merged stage. Requires the merged stage to be
1858        // loadable, so publish first (idempotent, content-addressed).
1859        self.publish(&merged_stage)?;
1860        let mut delta = std::collections::BTreeMap::new();
1861        delta.insert(sig_id.to_string(), Some(new_stage_id.clone()));
1862        match self.typecheck_merge_projection(dst_branch, &delta) {
1863            Ok(()) => Ok(Some(new_stage_id)),
1864            // Composes syntactically, not by type → still a conflict.
1865            Err(StoreError::TypeError(_)) => Ok(None),
1866            Err(e) => Err(e),
1867        }
1868    }
1869
1870    /// #836 G3: assemble everything a regenerator needs to *replay* an
1871    /// op — re-derive the change from its recorded cause. Returns the
1872    /// op's recorded intent (prompt / model / session), the target sig
1873    /// and the stage id it produced, and the program the change was
1874    /// made against (the parent state, rendered to source). An external
1875    /// harness feeds the prompt + parent program to the recorded model,
1876    /// then hands the regenerated stage back to [`Self::replay_compare`]
1877    /// (lex owns the deterministic comparison; the model call is the
1878    /// harness's, matching the rest of the architecture).
1879    ///
1880    /// Errors with `UnknownOp` if the op_id is unknown, or
1881    /// `InvalidTransition` if the op didn't produce a stage (a removal /
1882    /// import / merge has nothing to regenerate).
1883    pub fn replay_request(&self, op_id: &str) -> Result<ReplayRequest, StoreError> {
1884        let log = lex_vcs::OpLog::open(self.root())?;
1885        let record = log
1886            .get(&op_id.to_string())?
1887            .ok_or_else(|| StoreError::UnknownOp(op_id.to_string()))?;
1888        let (target_sig, expected_stage_id) = produced_sig_stage(&record.produces)
1889            .ok_or_else(|| StoreError::InvalidTransition(format!("op {op_id} produced no stage to replay")))?;
1890
1891        let (prompt, model, session_id) = match &record.op.intent_id {
1892            Some(id) => {
1893                let intents = lex_vcs::IntentLog::open(self.root())?;
1894                match intents.get(id)? {
1895                    Some(i) => (Some(i.prompt), Some(model_label(&i.model)), Some(i.session_id)),
1896                    None => (None, None, None),
1897                }
1898            }
1899            None => (None, None, None),
1900        };
1901
1902        // The program the op was applied against: the head state at its
1903        // (first) parent, rendered with the canonical printer. A root
1904        // op has no parent → empty program.
1905        let parent_program = match record.op.parents.first() {
1906            Some(parent) => self.program_source_at_op(parent)?,
1907            None => String::new(),
1908        };
1909
1910        // The target function's name + signature, from the recorded
1911        // stage — a regenerator needs the interface, not just the hash.
1912        let (target_name, target_signature) = match self.get_ast(&expected_stage_id) {
1913            Ok(lex_ast::Stage::FnDecl(fd)) => {
1914                (Some(fd.name.clone()), Some(lex_vcs::render_signature(&fd)))
1915            }
1916            _ => (None, None),
1917        };
1918
1919        Ok(ReplayRequest {
1920            op_id: op_id.to_string(),
1921            target_sig,
1922            target_name,
1923            target_signature,
1924            expected_stage_id,
1925            prompt,
1926            model,
1927            session_id,
1928            parent_program,
1929        })
1930    }
1931
1932    /// #836 G3: compare a regenerated `candidate` against what the op
1933    /// recorded producing, and emit the `Replay` attestation. The
1934    /// reproducibility claim made concrete — a faithful regeneration of
1935    /// the same function from the same cause yields the same
1936    /// content-addressed stage id.
1937    ///
1938    /// `reproduced` is true iff the candidate is the same sig *and* the
1939    /// same stage id the op recorded. A candidate for a different sig
1940    /// counts as "not reproduced" (`produced_stage_id: None`) rather
1941    /// than an error — it's a legitimate, if negative, replay result.
1942    /// The attestation is addressed to the op's recorded stage, so
1943    /// `list_for_stage` surfaces it alongside the TypeCheck/Examples
1944    /// evidence.
1945    pub fn replay_compare(
1946        &self,
1947        op_id: &str,
1948        candidate: &Stage,
1949    ) -> Result<ReplayOutcome, StoreError> {
1950        let (target_sig, expected_stage_id) = self.replay_target(op_id)?;
1951        let cand_sig = lex_ast::sig_id(candidate);
1952        let cand_stage = stage_id(candidate);
1953        let produced_stage_id = match (cand_sig.as_deref(), &cand_stage) {
1954            // Same function regenerated: the produced stage is
1955            // whatever it content-addresses to.
1956            (Some(s), Some(st)) if s == target_sig => Some(st.clone()),
1957            // A different sig (or an unhashable stage) isn't a
1958            // regeneration of this op's change.
1959            _ => None,
1960        };
1961        let reproduced = produced_stage_id.as_deref() == Some(expected_stage_id.as_str());
1962        let detail = if reproduced {
1963            None
1964        } else {
1965            Some("regeneration did not reproduce the recorded stage".to_string())
1966        };
1967        self.emit_replay(op_id, &expected_stage_id, produced_stage_id, reproduced, None, detail)
1968    }
1969
1970    /// Record a *negative* replay result for a regeneration that never
1971    /// yielded a comparable stage — the output didn't parse, or didn't
1972    /// define the target sig (#836 G3). Emits a `Replay { reproduced:
1973    /// false, produced_stage_id: None }` attestation with `reason` in
1974    /// its `Failed` detail, so an automated `lex op replay` run always
1975    /// records a verdict rather than aborting. `reason` is caller-supplied
1976    /// (e.g. "regenerated source did not parse").
1977    pub fn replay_record_miss(&self, op_id: &str, reason: &str) -> Result<ReplayOutcome, StoreError> {
1978        let (_target_sig, expected_stage_id) = self.replay_target(op_id)?;
1979        self.emit_replay(op_id, &expected_stage_id, None, false, None, Some(reason.to_string()))
1980    }
1981
1982    /// `(target_sig, expected_stage_id)` for a replayable op, or an
1983    /// error if the op is unknown or produced no stage.
1984    fn replay_target(&self, op_id: &str) -> Result<(String, String), StoreError> {
1985        let log = lex_vcs::OpLog::open(self.root())?;
1986        let record = log
1987            .get(&op_id.to_string())?
1988            .ok_or_else(|| StoreError::UnknownOp(op_id.to_string()))?;
1989        produced_sig_stage(&record.produces)
1990            .ok_or_else(|| StoreError::InvalidTransition(format!("op {op_id} produced no stage to replay")))
1991    }
1992
1993    /// Record a replay verdict the caller has already decided — used by
1994    /// the CLI's behavioral tier, which does the (VM-backed) equivalence
1995    /// check the store deliberately can't. `expected_stage_id` is looked
1996    /// up from the op. Set `behavioral_samples` to `Some(n)` when the
1997    /// candidate reproduced *behaviorally* over `n` sampled inputs rather
1998    /// than by exact stage-id match; the attestation then records that
1999    /// weaker-but-real claim distinctly.
2000    pub fn replay_record(
2001        &self,
2002        op_id: &str,
2003        produced_stage_id: Option<String>,
2004        reproduced: bool,
2005        behavioral_samples: Option<usize>,
2006        fail_detail: Option<String>,
2007    ) -> Result<ReplayOutcome, StoreError> {
2008        let (_target_sig, expected_stage_id) = self.replay_target(op_id)?;
2009        self.emit_replay(op_id, &expected_stage_id, produced_stage_id, reproduced, behavioral_samples, fail_detail)
2010    }
2011
2012    /// Compute the exact-match verdict for a candidate *without* emitting
2013    /// an attestation — `(expected_stage_id, produced_stage_id, exact)`.
2014    /// Lets a caller (the CLI) fall back to a behavioral check on a valid
2015    /// but non-identical candidate and emit a single verdict, instead of
2016    /// [`Self::replay_compare`]'s emit-immediately shape.
2017    pub fn replay_stage_of(
2018        &self,
2019        op_id: &str,
2020        candidate: &Stage,
2021    ) -> Result<(String, Option<String>, bool), StoreError> {
2022        let (target_sig, expected_stage_id) = self.replay_target(op_id)?;
2023        let cand_sig = lex_ast::sig_id(candidate);
2024        let cand_stage = stage_id(candidate);
2025        let produced_stage_id = match (cand_sig.as_deref(), &cand_stage) {
2026            (Some(s), Some(st)) if s == target_sig => Some(st.clone()),
2027            _ => None,
2028        };
2029        let exact = produced_stage_id.as_deref() == Some(expected_stage_id.as_str());
2030        Ok((expected_stage_id, produced_stage_id, exact))
2031    }
2032
2033    /// Emit the `Replay` attestation and build the outcome. Shared by
2034    /// [`Self::replay_compare`], [`Self::replay_record_miss`], and
2035    /// [`Self::replay_record`].
2036    fn emit_replay(
2037        &self,
2038        op_id: &str,
2039        expected_stage_id: &str,
2040        produced_stage_id: Option<String>,
2041        reproduced: bool,
2042        behavioral_samples: Option<usize>,
2043        fail_detail: Option<String>,
2044    ) -> Result<ReplayOutcome, StoreError> {
2045        let model = {
2046            let log = lex_vcs::OpLog::open(self.root())?;
2047            match log.get(&op_id.to_string())?.and_then(|r| r.op.intent_id) {
2048                Some(id) => lex_vcs::IntentLog::open(self.root())?
2049                    .get(&id)?
2050                    .map(|i| model_label(&i.model)),
2051                None => None,
2052            }
2053        };
2054        let result = if reproduced {
2055            lex_vcs::AttestationResult::Passed
2056        } else {
2057            lex_vcs::AttestationResult::Failed {
2058                detail: fail_detail.unwrap_or_else(|| "not reproduced".into()),
2059            }
2060        };
2061        let attestation = lex_vcs::Attestation::new(
2062            expected_stage_id.to_string(),
2063            Some(op_id.to_string()),
2064            None,
2065            lex_vcs::AttestationKind::Replay {
2066                expected_stage_id: expected_stage_id.to_string(),
2067                produced_stage_id: produced_stage_id.clone(),
2068                reproduced,
2069                behavioral_samples,
2070                model,
2071            },
2072            result,
2073            replay_producer(),
2074            None,
2075        );
2076        let attestation_id = attestation.attestation_id.clone();
2077        self.attestation_log()?.put(&attestation)?;
2078        Ok(ReplayOutcome {
2079            op_id: op_id.to_string(),
2080            expected_stage_id: expected_stage_id.to_string(),
2081            produced_stage_id,
2082            reproduced,
2083            behavioral_samples,
2084            attestation_id,
2085        })
2086    }
2087
2088    /// The program at an op (that op and all its ancestors applied), as
2089    /// canonical stages. The behavioral replay tier needs the whole
2090    /// program — a regenerated function may call helpers from its parent
2091    /// state, so it can only be run in context. Exposed for the CLI's
2092    /// equivalence check; `op_id` may be any op in the log.
2093    pub fn program_stages_at_op(&self, op_id: &str) -> Result<Vec<Stage>, StoreError> {
2094        let oid: lex_vcs::OpId = op_id.to_string();
2095        let log = lex_vcs::OpLog::open(self.root())?;
2096        let mut map: std::collections::BTreeMap<String, String> = std::collections::BTreeMap::new();
2097        for rec in log.walk_forward(&oid, None)? {
2098            crate::branches::apply_transition(&mut map, &rec.produces);
2099        }
2100        let pairs: Vec<(String, String)> = map.into_iter().collect();
2101        let stages: Vec<Stage> =
2102            self.get_asts_for_sigs_bulk(&pairs).into_iter().collect::<Result<_, _>>()?;
2103        Ok(stages)
2104    }
2105
2106    /// The program at an op, rendered to source. Used to give a replay
2107    /// regenerator the context the change was made against.
2108    fn program_source_at_op(&self, op_id: &lex_vcs::OpId) -> Result<String, StoreError> {
2109        Ok(lex_ast::print_stages(&self.program_stages_at_op(op_id)?))
2110    }
2111
2112    /// Open the attestation log rooted at this store. The log lives
2113    /// under `<root>/attestations/`; opening is idempotent and cheap
2114    /// (`fs::create_dir_all`). Exposed publicly so consumers — `lex
2115    /// blame --with-evidence`, `GET /v1/stage/<id>/attestations` —
2116    /// can read what the store gate emitted without round-tripping
2117    /// through this crate's API surface.
2118    /// Recompute a producer's trust score from its recent
2119    /// attestation history and emit a fresh `ProducerTrust`
2120    /// attestation (#293). Score = `passed / (passed + failed
2121    /// + inconclusive)` over the last `window` attestations
2122    /// produced by `tool_id`, expressed in thousandths
2123    /// (`0..=1000`).
2124    ///
2125    /// Refuses to grant trust when the tool has an active
2126    /// `ProducerBlock` — the block wins as a hard veto. Returns
2127    /// `Ok(None)` for "no attestations to score" (a brand-new
2128    /// producer); the caller can choose how to handle it
2129    /// (typically: skip the publish until evidence accrues).
2130    ///
2131    /// `granted_by` is the identity of the actor running the
2132    /// recompute (typically the human admin, or "lex-ci-bot"
2133    /// for an automated nightly).
2134    pub fn recompute_producer_trust(
2135        &self,
2136        tool_id: &str,
2137        window: usize,
2138        granted_by: &str,
2139    ) -> Result<Option<lex_vcs::AttestationId>, StoreError> {
2140        let log = self.attestation_log()?;
2141        let all = log.list_all()?;
2142        // Hard veto: don't grant trust to a blocked tool.
2143        if lex_vcs::active_producer_block(&all, tool_id).is_some() {
2144            return Err(StoreError::InvalidTransition(format!(
2145                "cannot recompute trust for `{tool_id}` — \
2146                 producer is currently blocked"
2147            )));
2148        }
2149        // Filter to attestations from this tool, newest-first by
2150        // timestamp, then take the window.
2151        let mut from_tool: Vec<&lex_vcs::Attestation> = all
2152            .iter()
2153            .filter(|a| a.produced_by.tool == tool_id)
2154            // Ignore self-referential trust attestations (we're
2155            // scoring evidence, not previous trust statements).
2156            .filter(|a| {
2157                !matches!(
2158                    a.kind,
2159                    lex_vcs::AttestationKind::ProducerTrust { .. }
2160                        | lex_vcs::AttestationKind::TrustWaived { .. }
2161                )
2162            })
2163            .collect();
2164        from_tool.sort_by_key(|a| std::cmp::Reverse(a.timestamp));
2165        from_tool.truncate(window);
2166        if from_tool.is_empty() {
2167            return Ok(None);
2168        }
2169        let (mut passed, mut total) = (0u64, 0u64);
2170        for a in &from_tool {
2171            total += 1;
2172            if matches!(a.result, lex_vcs::AttestationResult::Passed) {
2173                passed += 1;
2174            }
2175        }
2176        let score = if total == 0 {
2177            0
2178        } else {
2179            let raw = (passed as f64) * 1000.0 / (total as f64);
2180            raw.round().clamp(0.0, 1000.0) as u32
2181        };
2182        let head_op = self
2183            .list_branches()?
2184            .into_iter()
2185            .find_map(|b| self.get_branch(&b).ok().flatten().and_then(|x| x.head_op))
2186            .unwrap_or_else(|| "fresh".into());
2187        let evidence = format!(
2188            "window={window}, sample={}, head_op={head_op:.16}",
2189            from_tool.len()
2190        );
2191        let attestation = lex_vcs::Attestation::new(
2192            tool_id.to_string(),
2193            None,
2194            None,
2195            lex_vcs::AttestationKind::ProducerTrust {
2196                tool_id: tool_id.into(),
2197                score_thousandths: score,
2198                evidence,
2199                granted_by: granted_by.into(),
2200            },
2201            lex_vcs::AttestationResult::Passed,
2202            producer_trust_producer(),
2203            None,
2204        );
2205        let id = attestation.attestation_id.clone();
2206        log.put(&attestation)?;
2207        Ok(Some(id))
2208    }
2209
2210    /// The latest live `ProducerTrust` score (thousandths, `0..=1000`) for
2211    /// every producer that currently has trust: the newest score per tool by
2212    /// timestamp, excluding any tool under an active `ProducerBlock` (a block
2213    /// is a hard veto over trust, matching `recompute_producer_trust`).
2214    ///
2215    /// Used to export a capsule trusted-keys keyring from *earned* trust — the
2216    /// producer id doubles as the publisher's signing key downstream, so this
2217    /// turns track record into the allowlist `capsule install` consumes.
2218    pub fn live_producer_trust_scores(
2219        &self,
2220    ) -> Result<std::collections::BTreeMap<String, u32>, StoreError> {
2221        let log = self.attestation_log()?;
2222        let all = log.list_all()?;
2223        // Newest score per tool.
2224        let mut latest: std::collections::BTreeMap<String, (u64, u32)> =
2225            std::collections::BTreeMap::new();
2226        for a in &all {
2227            if let lex_vcs::AttestationKind::ProducerTrust {
2228                tool_id,
2229                score_thousandths,
2230                ..
2231            } = &a.kind
2232            {
2233                let entry = latest.entry(tool_id.clone()).or_insert((0, 0));
2234                if a.timestamp >= entry.0 {
2235                    *entry = (a.timestamp, *score_thousandths);
2236                }
2237            }
2238        }
2239        // Drop blocked producers; a block vetoes trust.
2240        let mut scores = std::collections::BTreeMap::new();
2241        for (tool, (_, score)) in latest {
2242            if lex_vcs::active_producer_block(&all, &tool).is_some() {
2243                continue;
2244            }
2245            scores.insert(tool, score);
2246        }
2247        Ok(scores)
2248    }
2249
2250    pub fn attestation_log(&self) -> Result<lex_vcs::AttestationLog, StoreError> {
2251        Ok(lex_vcs::AttestationLog::open(self.root())?)
2252    }
2253
2254    /// Emit one `TypeCheck::Passed` attestation per stage produced by
2255    /// a successful gated apply. Idempotent on `attestation_id` —
2256    /// re-running the same gate run dedups via content addressing.
2257    ///
2258    /// Failure modes: `io::Error` from the attestation log (disk
2259    /// full, perms). The op has already landed by the time this
2260    /// runs; an error here means the op is durable but the evidence
2261    /// is missing. We propagate so the caller sees the partial
2262    /// state rather than silently swallowing — re-attesting the
2263    /// same op against the same op_id is idempotent (content
2264    /// addressing) so a retry is safe once the underlying issue is
2265    /// fixed.
2266    fn record_typecheck_passed(
2267        &self,
2268        stage_ids: &[String],
2269        op_id: &lex_vcs::OpId,
2270    ) -> Result<(), StoreError> {
2271        if stage_ids.is_empty() {
2272            return Ok(());
2273        }
2274        let log = self.attestation_log()?;
2275        for stage_id in stage_ids {
2276            let attestation = lex_vcs::Attestation::new(
2277                stage_id.clone(),
2278                Some(op_id.clone()),
2279                None,
2280                lex_vcs::AttestationKind::TypeCheck,
2281                lex_vcs::AttestationResult::Passed,
2282                typecheck_producer(),
2283                None,
2284            );
2285            log.put(&attestation)?;
2286        }
2287        Ok(())
2288    }
2289
2290    /// The hosted CI runner (#93): independently re-run the write-time
2291    /// type-check gate on a branch head and record the verdict as a
2292    /// `lex-hub-ci`-produced `TypeCheck` attestation for the stages the
2293    /// advance introduced. Called after an `op push` fast-forwards the
2294    /// head, so `require-attestation type_check` gates are backed by a
2295    /// producer that actually verified the code server-side, not by
2296    /// whatever attestation a client chose to attach. Does NOT move or
2297    /// roll back the head — the client's own always-valid-HEAD gate is
2298    /// what refuses a bad publish; this produces the trusted verdict on
2299    /// top of an already-committed advance (so a client that bypassed
2300    /// its gate is caught by a `TypeCheck::Failed` from `lex-hub-ci`).
2301    ///
2302    /// `from_head` is the branch head *before* the advance; the ops
2303    /// between it and `to_head` are the ones whose stages get attested.
2304    /// Idempotent: attestations are content-addressed, so re-verifying
2305    /// the same head is a no-op.
2306    pub fn verify_head_and_attest(
2307        &self,
2308        branch: &str,
2309        from_head: Option<&str>,
2310        to_head: &str,
2311    ) -> Result<HubCiVerdict, StoreError> {
2312        // Reconstruct the program at the new head and re-check it.
2313        let head = self.branch_head(branch)?;
2314        let pairs: Vec<(String, String)> = head.into_iter().collect();
2315        let decls: Vec<Stage> =
2316            self.get_asts_for_sigs_bulk(&pairs).into_iter().collect::<Result<_, _>>()?;
2317        let checked_stages = decls.len();
2318        // #930: the SigId→stage map holds only fn/type declarations — the
2319        // head's `import` edges are AddImport ops, absent here. Reconstruct
2320        // them so a non-inlined head's `<alias>.name` references bind: the
2321        // resolver scans these imports to resolve each dependency, and the
2322        // checker's Pass 1 binds the alias to the resolved module. (Without
2323        // this the alias is unbound and the head fails as `unknown_identifier`,
2324        // even with the dependency correctly resolved.)
2325        let head_imports = crate::render::package_head_at_op(self, to_head)
2326            .map(|ph| {
2327                ph.flat_imports
2328                    .into_iter()
2329                    .map(|(reference, alias)| {
2330                        Stage::Import(lex_ast::Import { reference, alias })
2331                    })
2332                    .collect::<Vec<_>>()
2333            })
2334            .unwrap_or_default();
2335        let mut stages = head_imports;
2336        stages.extend(decls);
2337        // #930: the hub gate resolves this head's external dependencies from
2338        // the lock committed with `to_head` (via the installed cross-store
2339        // resolver); empty when none is installed or the head is inlined.
2340        let modules = self.resolved_modules(&stages, Some(to_head));
2341        let result = match lex_types::check_program_with_modules(&stages, &modules) {
2342            Ok(_) => lex_vcs::AttestationResult::Passed,
2343            Err(errors) => lex_vcs::AttestationResult::Failed {
2344                detail: serde_json::to_string(&errors).unwrap_or_else(|_| "type errors".into()),
2345            },
2346        };
2347        let passed = matches!(result, lex_vcs::AttestationResult::Passed);
2348
2349        // Stages introduced by THIS advance (from_head exclusive → to_head).
2350        let log = lex_vcs::OpLog::open(self.root())?;
2351        let to = to_head.to_string();
2352        let records = match from_head {
2353            Some(f) => log
2354                .walk_forward_since(&to, &f.to_string())?
2355                .unwrap_or_else(|| log.walk_forward(&to, None).unwrap_or_default()),
2356            None => log.walk_forward(&to, None)?,
2357        };
2358        let mut introduced: Vec<String> = Vec::new();
2359        for rec in &records {
2360            introduced.extend(attestable_stage_ids(&rec.produces));
2361        }
2362
2363        let alog = self.attestation_log()?;
2364        for sid in &introduced {
2365            let att = lex_vcs::Attestation::new(
2366                sid.clone(),
2367                Some(to_head.to_string()),
2368                None,
2369                lex_vcs::AttestationKind::TypeCheck,
2370                result.clone(),
2371                hub_ci_producer(),
2372                None,
2373            );
2374            alog.put(&att)?;
2375        }
2376
2377        let detail = match &result {
2378            lex_vcs::AttestationResult::Failed { detail } => Some(detail.clone()),
2379            _ => None,
2380        };
2381        Ok(HubCiVerdict { passed, checked_stages, attested_stages: introduced.len(), detail })
2382    }
2383
2384    /// Emit an `Examples::Passed` attestation for a published stage
2385    /// whose behavioral `examples {}` block was run and passed (#835,
2386    /// Tier 1). Mirrors [`Self::record_typecheck_passed`]. The
2387    /// behavioral run itself happens one layer up (lex-api / lex-cli)
2388    /// because it needs the bytecode compiler + VM, which this crate
2389    /// deliberately doesn't depend on; the store only records the
2390    /// verdict. `file_hash` uses the stage id — the stage fully
2391    /// determines its own examples.
2392    pub fn record_examples_passed(
2393        &self,
2394        stage_id: &str,
2395        op_id: &lex_vcs::OpId,
2396        count: usize,
2397    ) -> Result<(), StoreError> {
2398        let log = self.attestation_log()?;
2399        let attestation = lex_vcs::Attestation::new(
2400            stage_id.to_string(),
2401            Some(op_id.clone()),
2402            None,
2403            lex_vcs::AttestationKind::Examples { file_hash: stage_id.to_string(), count },
2404            lex_vcs::AttestationResult::Passed,
2405            examples_producer(),
2406            None,
2407        );
2408        log.put(&attestation)?;
2409        Ok(())
2410    }
2411
2412    /// Record a structured `Review` verdict on a stage (#836 G4).
2413    /// The verdict maps onto the attestation `result` so existing
2414    /// result-based tooling reads it: Approve->Passed,
2415    /// Reject->Failed, RequestChanges->Inconclusive.
2416    pub fn record_review(
2417        &self,
2418        stage_id: &str,
2419        op_id: Option<lex_vcs::OpId>,
2420        reviewer: &str,
2421        verdict: lex_vcs::ReviewVerdict,
2422        notes: Option<String>,
2423    ) -> Result<lex_vcs::AttestationId, StoreError> {
2424        let result = match verdict {
2425            lex_vcs::ReviewVerdict::Approve => lex_vcs::AttestationResult::Passed,
2426            lex_vcs::ReviewVerdict::Reject => lex_vcs::AttestationResult::Failed {
2427                detail: notes.clone().unwrap_or_else(|| "rejected".into()),
2428            },
2429            lex_vcs::ReviewVerdict::RequestChanges => lex_vcs::AttestationResult::Inconclusive {
2430                detail: notes.clone().unwrap_or_else(|| "changes requested".into()),
2431            },
2432        };
2433        let att = lex_vcs::Attestation::new(
2434            stage_id.to_string(),
2435            op_id,
2436            None,
2437            lex_vcs::AttestationKind::Review { reviewer: reviewer.to_string(), verdict, notes },
2438            result,
2439            review_producer(reviewer),
2440            None,
2441        );
2442        let id = att.attestation_id.clone();
2443        self.attestation_log()?.put(&att)?;
2444        Ok(id)
2445    }
2446
2447    /// The latest `Review` verdict recorded on a stage, if any
2448    /// (#836 G4). "Latest" is by attestation timestamp; ties keep the
2449    /// last one seen. Used by `promote_candidate` to honor a standing
2450    /// Reject.
2451    pub fn latest_review_verdict(
2452        &self,
2453        stage_id: &str,
2454    ) -> Result<Option<lex_vcs::ReviewVerdict>, StoreError> {
2455        let log = self.attestation_log()?;
2456        let mut latest: Option<(u64, lex_vcs::ReviewVerdict)> = None;
2457        for a in log.list_for_stage(&stage_id.to_string())? {
2458            if let lex_vcs::AttestationKind::Review { verdict, .. } = a.kind {
2459                if latest.as_ref().map(|(t, _)| a.timestamp >= *t).unwrap_or(true) {
2460                    latest = Some((a.timestamp, verdict));
2461                }
2462            }
2463        }
2464        Ok(latest.map(|(_, v)| v))
2465    }
2466
2467    /// Consult `policy.session_budgets` for the op's session
2468    /// (resolved via `op.intent_id → Intent.session_id`) and
2469    /// refuse if applying would push the session's monotonic spend
2470    /// over the configured cap (#292 slice 3).
2471    ///
2472    /// Ops without an `intent_id`, or whose intent has no
2473    /// configured cap, return Ok without any disk read.
2474    fn check_session_budget(&self, op: &lex_vcs::Operation) -> Result<(), StoreError> {
2475        let Some(intent_id) = op.intent_id.as_deref() else {
2476            return Ok(());
2477        };
2478        let intent_log = lex_vcs::IntentLog::open(self.root())?;
2479        let Some(intent) = intent_log.get(&intent_id.to_string())? else {
2480            // Dangling intent — treat as "no session" and let it
2481            // sail through. Slice 1's ledger already documents
2482            // this as graceful-degradation semantics.
2483            return Ok(());
2484        };
2485        let policy = crate::policy::load(self.root())?.unwrap_or_default();
2486        let Some(cap) = policy.session_budgets.cap_for(&intent.session_id) else {
2487            return Ok(());
2488        };
2489        // Recompute the session's current spend + the contribution
2490        // from this op. Re-running the ledger walk on every gated
2491        // op is O(branch history); see #292 slice 1's note about
2492        // a future on-disk cache.
2493        let current = self.session_budget(&intent.session_id)?;
2494        let increment = crate::budget::monotonic_spend_of(&op.kind);
2495        let spent_after = current.spent.saturating_add(increment);
2496        if spent_after > cap {
2497            return Err(StoreError::BudgetExceeded {
2498                session_id: intent.session_id,
2499                cap,
2500                spent_after,
2501            });
2502        }
2503        Ok(())
2504    }
2505
2506    /// Emit `RepairHint` attestations for a TypeError-rejected op
2507    /// (#281). One per candidate stage in the transition. The hint
2508    /// records the *would-be* op_id (deterministic, content-
2509    /// addressed even though the op record was never persisted)
2510    /// and the structured errors.
2511    ///
2512    /// #306 slice 3: `suggested_transform` is populated from the
2513    /// static (rule_tag → likely_transform) table for the *first*
2514    /// error in the batch. The LLM-driven `lex repair --apply`
2515    /// flow can still overwrite this with a higher-quality
2516    /// suggestion; the static value is the floor, not the ceiling.
2517    ///
2518    /// Best-effort: a write failure here is swallowed by the
2519    /// caller (the original `TypeError` is the load-bearing
2520    /// signal; missing the hint is recoverable on a retry).
2521    fn record_repair_hint(
2522        &self,
2523        stage_ids: &[String],
2524        failed_op_id: &lex_vcs::OpId,
2525        errors: &[lex_types::TypeError],
2526    ) -> Result<(), StoreError> {
2527        if stage_ids.is_empty() {
2528            return Ok(());
2529        }
2530        let errors_json = serde_json::to_value(errors).map_err(StoreError::Serde)?;
2531        // #306 slice 3: look up the static suggested_transform for
2532        // the first error's rule_tag. Multiple errors per op are
2533        // possible — when they fire in lockstep (e.g. one bad let
2534        // binding propagates to several use sites), the first
2535        // error's rule_tag is usually the load-bearing one to fix.
2536        let suggested_transform = errors
2537            .first()
2538            .and_then(|e| lex_types::suggested_transform_for(e.rule_tag()));
2539        let log = self.attestation_log()?;
2540        for stage_id in stage_ids {
2541            let attestation = lex_vcs::Attestation::new(
2542                stage_id.clone(),
2543                None, // the failed op was never persisted; not the
2544                // attestation's op_id (which is for a
2545                // *successful* op).
2546                None,
2547                lex_vcs::AttestationKind::RepairHint {
2548                    failed_op_id: failed_op_id.clone(),
2549                    errors: errors_json.clone(),
2550                    suggested_transform: suggested_transform.clone(),
2551                },
2552                lex_vcs::AttestationResult::Failed {
2553                    detail: format!(
2554                        "op {} rejected: {} type error(s)",
2555                        failed_op_id,
2556                        errors.len()
2557                    ),
2558                },
2559                repair_hint_producer(),
2560                None,
2561            );
2562            log.put(&attestation)?;
2563        }
2564        Ok(())
2565    }
2566
2567    /// Emit `Trace` attestations linking an already-committed `op`
2568    /// to the run that produced it (#257). One attestation per
2569    /// produced stage (matching the `TypeCheck` emission contract
2570    /// — see [`Self::apply_operation_checked`]) with
2571    /// `op_id: Some(op_id)` set, so `lex trace --op <op_id>`
2572    /// surfaces the run.
2573    ///
2574    /// Returns the number of attestations emitted (zero for ops
2575    /// that produce no attestable stage, e.g. `Remove` /
2576    /// `ImportOnly`).
2577    ///
2578    /// Idempotent: re-emitting for the same
2579    /// `(run_id, root_target, op_id, stage_id, producer, result)`
2580    /// tuple dedups via content addressing.
2581    ///
2582    /// `op_id` must already exist in the op log — an unknown op
2583    /// surfaces as `StoreError::UnknownOp`.
2584    pub fn record_op_trace(
2585        &self,
2586        run_id: &str,
2587        root_target: &str,
2588        op_id: &lex_vcs::OpId,
2589        result: lex_vcs::AttestationResult,
2590        producer: lex_vcs::ProducerDescriptor,
2591    ) -> Result<usize, StoreError> {
2592        let log = lex_vcs::OpLog::open(self.root())?;
2593        let rec = log
2594            .get(op_id)?
2595            .ok_or_else(|| StoreError::UnknownOp(op_id.clone()))?;
2596        let stage_ids = attestable_stage_ids(&rec.produces);
2597        if stage_ids.is_empty() {
2598            return Ok(0);
2599        }
2600        let attlog = self.attestation_log()?;
2601        let mut emitted = 0;
2602        for stage_id in stage_ids {
2603            let attestation = lex_vcs::Attestation::new(
2604                stage_id,
2605                Some(op_id.clone()),
2606                None,
2607                lex_vcs::AttestationKind::Trace {
2608                    run_id: run_id.into(),
2609                    root_target: root_target.into(),
2610                },
2611                result.clone(),
2612                producer.clone(),
2613                None,
2614            );
2615            attlog.put(&attestation)?;
2616            emitted += 1;
2617        }
2618        Ok(emitted)
2619    }
2620
2621    /// Walk `ops_since(branch_head, base)` and emit per-stage
2622    /// `Trace` attestations for each new op, linking them to the
2623    /// run that produced them (#257). Used by `lex run --trace`
2624    /// after the VM exits: snapshot `base = branch_head` before
2625    /// the run, then call this with the post-run head.
2626    ///
2627    /// `base = None` means "every op currently reachable from the
2628    /// branch head" — generally not what you want for a single
2629    /// run; pass the pre-run head.
2630    ///
2631    /// Returns the total number of attestations emitted across
2632    /// every new op. Zero is the common case (the run committed no
2633    /// ops).
2634    ///
2635    /// Idempotent on the per-op level via [`Self::record_op_trace`].
2636    pub fn record_run_committed_ops_since(
2637        &self,
2638        run_id: &str,
2639        root_target: &str,
2640        branch: &str,
2641        base: Option<&lex_vcs::OpId>,
2642        result: lex_vcs::AttestationResult,
2643        producer: lex_vcs::ProducerDescriptor,
2644    ) -> Result<usize, StoreError> {
2645        let head = match self.get_branch(branch)?.and_then(|b| b.head_op) {
2646            Some(h) => h,
2647            None => return Ok(0),
2648        };
2649        let log = lex_vcs::OpLog::open(self.root())?;
2650        let new_ops = log.ops_since(&head, base)?;
2651        let mut total = 0;
2652        for rec in new_ops {
2653            total += self.record_op_trace(
2654                run_id,
2655                root_target,
2656                &rec.op_id,
2657                result.clone(),
2658                producer.clone(),
2659            )?;
2660        }
2661        Ok(total)
2662    }
2663
2664    /// Apply a typed `ReplaceMatchArm` transform (#280) and emit a
2665    /// `OperationKind::ReplaceMatchArm` op that records the
2666    /// semantic shape of the edit, not just the byte effect.
2667    ///
2668    /// Steps:
2669    ///   1. Load the source stage's canonical bytes (delta-aware).
2670    ///   2. Run [`lex_ast::replace_match_arm`] to produce the new
2671    ///      `Stage`. Pure function, no I/O.
2672    ///   3. Publish the new stage. Idempotent on the
2673    ///      content-addressed `to_stage_id`.
2674    ///   4. Assemble the candidate program (every active stage on
2675    ///      the branch, with the rewritten one swapped in) and call
2676    ///      [`Self::apply_operation_checked`] — re-typechecks and
2677    ///      runs every existing gate (TypeCheck attestation,
2678    ///      required_attestations, producer-block walk-back).
2679    ///
2680    /// Failure modes:
2681    ///   * [`StoreError::TransformError`] — transform didn't apply.
2682    ///     The branch is unchanged; no stage published.
2683    ///   * [`StoreError::TypeError`] — transform produced an
2684    ///     ill-typed program. The new stage is on disk (idempotent
2685    ///     on its content hash) but the branch is unchanged. Same
2686    ///     "publish without advance" semantics as #245.
2687    ///   * Everything else from `apply_operation_checked`.
2688    pub fn apply_replace_match_arm(
2689        &self,
2690        branch: &str,
2691        from_stage_id: &str,
2692        match_node: &lex_ast::NodeId,
2693        arm_index: usize,
2694        new_body: lex_ast::CExpr,
2695    ) -> Result<lex_vcs::OpId, StoreError> {
2696        let from_stage = self.get_ast(from_stage_id)?;
2697        let new_stage = lex_ast::replace_match_arm(&from_stage, match_node, arm_index, new_body)
2698            .map_err(StoreError::TransformError)?;
2699        let sig = lex_ast::sig_id(&from_stage).ok_or(StoreError::CannotPublishImport)?;
2700        let to_stage_id = self.publish(&new_stage)?;
2701        if to_stage_id == from_stage_id {
2702            // No-op transform — the new body was structurally
2703            // identical to the old. Refuse rather than advancing
2704            // the branch with an empty edit.
2705            return Err(StoreError::InvalidTransition(format!(
2706                "replace_match_arm produced the same stage_id `{from_stage_id}`"
2707            )));
2708        }
2709
2710        // Assemble the candidate program: every active stage on
2711        // the branch, with `from_stage_id` swapped for `new_stage`.
2712        let head = self.branch_head(branch)?;
2713        let mut candidate: Vec<lex_ast::Stage> = Vec::with_capacity(head.len());
2714        for (other_sig, other_stage_id) in &head {
2715            if other_sig == &sig {
2716                candidate.push(new_stage.clone());
2717            } else {
2718                candidate.push(self.get_ast(other_stage_id)?);
2719            }
2720        }
2721        // If the source sig isn't on the current branch head, the
2722        // transform is operating on a stage that hasn't been added
2723        // yet — refuse rather than risking a candidate program
2724        // that doesn't reflect the branch's actual state.
2725        if !head.contains_key(&sig) {
2726            return Err(StoreError::InvalidTransition(format!(
2727                "sig `{sig}` not on branch `{branch}`'s head"
2728            )));
2729        }
2730
2731        // #247: budget delta captured for `lex op log --budget-drift`.
2732        let from_budget = budget_of_stage(&from_stage);
2733        let to_budget = budget_of_stage(&new_stage);
2734
2735        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
2736        let kind = lex_vcs::OperationKind::ReplaceMatchArm {
2737            sig_id: sig.clone(),
2738            from_stage_id: from_stage_id.to_string(),
2739            to_stage_id: to_stage_id.clone(),
2740            match_node: match_node.as_str().to_string(),
2741            arm_index,
2742            from_budget,
2743            to_budget,
2744        };
2745        let transition = lex_vcs::StageTransition::Replace {
2746            sig_id: sig.clone(),
2747            from: from_stage_id.to_string(),
2748            to: to_stage_id.clone(),
2749        };
2750        let op = lex_vcs::Operation::new(kind, head_now.into_iter().collect::<Vec<_>>());
2751        self.apply_operation_checked(branch, op, transition, &candidate)
2752    }
2753
2754    /// Apply a typed `RenameLocal` transform (#280) — rename a
2755    /// `let`-bound local within a fn body and emit a matching
2756    /// `OperationKind::RenameLocal`. Same end-to-end shape as
2757    /// [`Self::apply_replace_match_arm`]; see that method for the
2758    /// failure-mode taxonomy.
2759    pub fn apply_rename_local(
2760        &self,
2761        branch: &str,
2762        from_stage_id: &str,
2763        let_node: &lex_ast::NodeId,
2764        new_name: &str,
2765    ) -> Result<lex_vcs::OpId, StoreError> {
2766        let from_stage = self.get_ast(from_stage_id)?;
2767        // Read the old name before running the transform, so the
2768        // op log records the rename target rather than just the
2769        // new value.
2770        let old_name = read_let_name(&from_stage, let_node).map_err(StoreError::TransformError)?;
2771        let new_stage = lex_ast::rename_local(&from_stage, let_node, new_name)
2772            .map_err(StoreError::TransformError)?;
2773        let sig = lex_ast::sig_id(&from_stage).ok_or(StoreError::CannotPublishImport)?;
2774        let to_stage_id = self.publish(&new_stage)?;
2775        if to_stage_id == from_stage_id {
2776            return Err(StoreError::InvalidTransition(format!(
2777                "rename_local produced the same stage_id `{from_stage_id}`"
2778            )));
2779        }
2780        let head = self.branch_head(branch)?;
2781        let mut candidate: Vec<lex_ast::Stage> = Vec::with_capacity(head.len());
2782        for (other_sig, other_stage_id) in &head {
2783            if other_sig == &sig {
2784                candidate.push(new_stage.clone());
2785            } else {
2786                candidate.push(self.get_ast(other_stage_id)?);
2787            }
2788        }
2789        if !head.contains_key(&sig) {
2790            return Err(StoreError::InvalidTransition(format!(
2791                "sig `{sig}` not on branch `{branch}`'s head"
2792            )));
2793        }
2794        let from_budget = budget_of_stage(&from_stage);
2795        let to_budget = budget_of_stage(&new_stage);
2796        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
2797        let kind = lex_vcs::OperationKind::RenameLocal {
2798            sig_id: sig.clone(),
2799            from_stage_id: from_stage_id.to_string(),
2800            to_stage_id: to_stage_id.clone(),
2801            let_node: let_node.as_str().to_string(),
2802            old_name,
2803            new_name: new_name.to_string(),
2804            from_budget,
2805            to_budget,
2806        };
2807        let transition = lex_vcs::StageTransition::Replace {
2808            sig_id: sig.clone(),
2809            from: from_stage_id.to_string(),
2810            to: to_stage_id.clone(),
2811        };
2812        let op = lex_vcs::Operation::new(kind, head_now.into_iter().collect::<Vec<_>>());
2813        self.apply_operation_checked(branch, op, transition, &candidate)
2814    }
2815
2816    /// Apply a typed `InlineLet` transform (#280) — eliminate a
2817    /// `let x := v; body` by substituting `v` for every unshadowed
2818    /// `x` in `body`, then replacing the `Let` node with the
2819    /// substituted body. Same end-to-end shape as
2820    /// [`Self::apply_replace_match_arm`].
2821    pub fn apply_inline_let(
2822        &self,
2823        branch: &str,
2824        from_stage_id: &str,
2825        let_node: &lex_ast::NodeId,
2826    ) -> Result<lex_vcs::OpId, StoreError> {
2827        let from_stage = self.get_ast(from_stage_id)?;
2828        let binding_name =
2829            read_let_name(&from_stage, let_node).map_err(StoreError::TransformError)?;
2830        let new_stage =
2831            lex_ast::inline_let(&from_stage, let_node).map_err(StoreError::TransformError)?;
2832        let sig = lex_ast::sig_id(&from_stage).ok_or(StoreError::CannotPublishImport)?;
2833        let to_stage_id = self.publish(&new_stage)?;
2834        if to_stage_id == from_stage_id {
2835            return Err(StoreError::InvalidTransition(format!(
2836                "inline_let produced the same stage_id `{from_stage_id}`"
2837            )));
2838        }
2839        let head = self.branch_head(branch)?;
2840        let mut candidate: Vec<lex_ast::Stage> = Vec::with_capacity(head.len());
2841        for (other_sig, other_stage_id) in &head {
2842            if other_sig == &sig {
2843                candidate.push(new_stage.clone());
2844            } else {
2845                candidate.push(self.get_ast(other_stage_id)?);
2846            }
2847        }
2848        if !head.contains_key(&sig) {
2849            return Err(StoreError::InvalidTransition(format!(
2850                "sig `{sig}` not on branch `{branch}`'s head"
2851            )));
2852        }
2853        let from_budget = budget_of_stage(&from_stage);
2854        let to_budget = budget_of_stage(&new_stage);
2855        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
2856        let kind = lex_vcs::OperationKind::InlineLet {
2857            sig_id: sig.clone(),
2858            from_stage_id: from_stage_id.to_string(),
2859            to_stage_id: to_stage_id.clone(),
2860            let_node: let_node.as_str().to_string(),
2861            binding_name,
2862            from_budget,
2863            to_budget,
2864        };
2865        let transition = lex_vcs::StageTransition::Replace {
2866            sig_id: sig.clone(),
2867            from: from_stage_id.to_string(),
2868            to: to_stage_id.clone(),
2869        };
2870        let op = lex_vcs::Operation::new(kind, head_now.into_iter().collect::<Vec<_>>());
2871        self.apply_operation_checked(branch, op, transition, &candidate)
2872    }
2873
2874    /// Apply a typed `ExtractFunction` transform (#280 slice 4) —
2875    /// extract a sub-expression of `from_stage_id`'s body into a
2876    /// new top-level fn defined by `spec`, and emit two ops tied
2877    /// together by a shared synthetic Intent so `lex op log
2878    /// --intent <id>` groups them.
2879    ///
2880    /// The two ops:
2881    ///   1. `AddFunction { sig_id: <new_fn_sig>, stage_id: <new_fn_stage> }`
2882    ///   2. `ModifyBody { sig_id: <source_sig>, from_stage_id, to_stage_id: <modified> }`
2883    ///
2884    /// The shared Intent's prompt is structured (`extract_function:
2885    /// <new_fn_name>` plus the source identity) so downstream
2886    /// tooling can recover the typed-transform shape from the
2887    /// op-log + intent-log join.
2888    ///
2889    /// Returns `(add_fn_op_id, modify_body_op_id)`.
2890    pub fn apply_extract_function(
2891        &self,
2892        branch: &str,
2893        from_stage_id: &str,
2894        expr_node: &lex_ast::NodeId,
2895        spec: lex_ast::ExtractFnSpec,
2896    ) -> Result<(lex_vcs::OpId, lex_vcs::OpId), StoreError> {
2897        let from_stage = self.get_ast(from_stage_id)?;
2898        let new_fn_name = spec.name.clone();
2899        let (modified_stage, new_fn_stage) =
2900            lex_ast::extract_function(&from_stage, expr_node, spec)
2901                .map_err(StoreError::TransformError)?;
2902
2903        let source_sig = lex_ast::sig_id(&from_stage).ok_or(StoreError::CannotPublishImport)?;
2904        let new_fn_sig = lex_ast::sig_id(&new_fn_stage).ok_or(StoreError::CannotPublishImport)?;
2905        if source_sig == new_fn_sig {
2906            return Err(StoreError::InvalidTransition(format!(
2907                "extract_function produced a sig matching the source `{source_sig}`"
2908            )));
2909        }
2910        let new_fn_stage_id = self.publish(&new_fn_stage)?;
2911        let modified_stage_id = self.publish(&modified_stage)?;
2912        if modified_stage_id == from_stage_id {
2913            return Err(StoreError::InvalidTransition(format!(
2914                "extract_function produced the same stage_id `{from_stage_id}` for the source"
2915            )));
2916        }
2917
2918        let head = self.branch_head(branch)?;
2919        if !head.contains_key(&source_sig) {
2920            return Err(StoreError::InvalidTransition(format!(
2921                "sig `{source_sig}` not on branch `{branch}`'s head"
2922            )));
2923        }
2924
2925        // Synthesize an Intent linking the two ops. The session_id
2926        // / model fields here are not load-bearing — they exist to
2927        // make the IntentId content-addressed; downstream tooling
2928        // reads `prompt` to reconstruct the typed-transform shape.
2929        let intent = lex_vcs::Intent::new(
2930            format!(
2931                "[lex.transform.extract_function]\nnew_fn={new_fn_name}\nsource_sig={source_sig}\nfrom_stage={from_stage_id}\nexpr_node={node}",
2932                node = expr_node.as_str(),
2933            ),
2934            "lex-store::apply_extract_function",
2935            lex_vcs::ModelDescriptor {
2936                provider: "lex-store".into(),
2937                name: env!("CARGO_PKG_VERSION").into(),
2938                version: None,
2939            },
2940            None,
2941        );
2942        let intent_id = intent.intent_id.clone();
2943        lex_vcs::IntentLog::open(self.root())?.put(&intent)?;
2944
2945        // Step 1 — emit the AddFunction op for the new fn. Build
2946        // the candidate program by appending the new fn to every
2947        // stage on the current branch head.
2948        let new_fn_effects: std::collections::BTreeSet<String> = match &new_fn_stage {
2949            lex_ast::Stage::FnDecl(fd) => fd.effects.iter().map(|e| e.name.clone()).collect(),
2950            _ => Default::default(),
2951        };
2952        let new_fn_budget = budget_of_stage(&new_fn_stage);
2953        let mut candidate_with_new_fn: Vec<lex_ast::Stage> = Vec::with_capacity(head.len() + 1);
2954        for stage_id in head.values() {
2955            candidate_with_new_fn.push(self.get_ast(stage_id)?);
2956        }
2957        candidate_with_new_fn.push(new_fn_stage.clone());
2958        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
2959        let add_op = lex_vcs::Operation::new(
2960            lex_vcs::OperationKind::AddFunction {
2961                sig_id: new_fn_sig.clone(),
2962                stage_id: new_fn_stage_id.clone(),
2963                effects: new_fn_effects,
2964                budget_cost: new_fn_budget,
2965                // Single-op apply path — no package context here.
2966                in_file: None,
2967            },
2968            head_now.into_iter().collect::<Vec<_>>(),
2969        )
2970        .with_intent(intent_id.clone());
2971        let add_transition = lex_vcs::StageTransition::Create {
2972            sig_id: new_fn_sig.clone(),
2973            stage_id: new_fn_stage_id.clone(),
2974        };
2975        let add_op_id =
2976            self.apply_operation_checked(branch, add_op, add_transition, &candidate_with_new_fn)?;
2977
2978        // Step 2 — emit the ModifyBody op for the source. Build
2979        // the candidate program by replacing the source's stage
2980        // with `modified_stage` and keeping the new fn alongside.
2981        let from_budget = budget_of_stage(&from_stage);
2982        let to_budget = budget_of_stage(&modified_stage);
2983        let mut candidate_with_modified: Vec<lex_ast::Stage> = Vec::with_capacity(head.len() + 1);
2984        for (other_sig, other_stage_id) in &head {
2985            if other_sig == &source_sig {
2986                candidate_with_modified.push(modified_stage.clone());
2987            } else {
2988                candidate_with_modified.push(self.get_ast(other_stage_id)?);
2989            }
2990        }
2991        candidate_with_modified.push(new_fn_stage.clone());
2992        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
2993        let modify_op = lex_vcs::Operation::new(
2994            lex_vcs::OperationKind::ModifyBody {
2995                sig_id: source_sig.clone(),
2996                from_stage_id: from_stage_id.to_string(),
2997                to_stage_id: modified_stage_id.clone(),
2998                from_budget,
2999                to_budget,
3000            },
3001            head_now.into_iter().collect::<Vec<_>>(),
3002        )
3003        .with_intent(intent_id);
3004        let modify_transition = lex_vcs::StageTransition::Replace {
3005            sig_id: source_sig,
3006            from: from_stage_id.to_string(),
3007            to: modified_stage_id,
3008        };
3009        let modify_op_id = self.apply_operation_checked(
3010            branch,
3011            modify_op,
3012            modify_transition,
3013            &candidate_with_modified,
3014        )?;
3015
3016        Ok((add_op_id, modify_op_id))
3017    }
3018
3019    /// Propose a stage for `sig_id` without advancing the branch
3020    /// head (#294). Multiple agents can call this concurrently
3021    /// for the same sig — every call lands a fresh `Candidate`
3022    /// op chained off the current head_op. The branch head stays
3023    /// where it was; a later [`Self::promote_candidate`] picks
3024    /// the winner.
3025    ///
3026    /// The caller is responsible for typechecking `new_stage`
3027    /// against whatever program context they consider valid —
3028    /// `propose_candidate` doesn't run the gate. Type errors
3029    /// surface at promotion time, where the candidate is
3030    /// composed back into a candidate program via the standard
3031    /// `apply_operation_checked` path.
3032    ///
3033    /// The stage is published (idempotent on content hash). The
3034    /// `intent_id` is required so downstream consumers can
3035    /// distinguish proposals by author.
3036    pub fn propose_candidate(
3037        &self,
3038        branch: &str,
3039        new_stage: &lex_ast::Stage,
3040        intent_id: &lex_vcs::IntentId,
3041    ) -> Result<lex_vcs::OpId, StoreError> {
3042        let sig = lex_ast::sig_id(new_stage).ok_or(StoreError::CannotPublishImport)?;
3043        let stage_id = self.publish(new_stage)?;
3044        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
3045        let op = lex_vcs::Operation::new(
3046            lex_vcs::OperationKind::Candidate {
3047                sig_id: sig,
3048                stage_id,
3049            },
3050            head_now.into_iter().collect::<Vec<_>>(),
3051        )
3052        .with_intent(intent_id.clone());
3053        let transition = lex_vcs::StageTransition::ImportOnly;
3054        self.apply_operation(branch, op, transition)
3055    }
3056
3057    /// List every live `Candidate` op for `sig_id` — i.e. those
3058    /// not yet referenced by any `Promote` op (either as the
3059    /// winner or in the `supersedes` set). Used by `lex stage
3060    /// candidates`. Results are sorted by op_id for
3061    /// reproducibility.
3062    pub fn list_candidates(&self, sig_id: &str) -> Result<Vec<CandidateInfo>, StoreError> {
3063        let log = lex_vcs::OpLog::open(self.root())?;
3064        let all = log.list_all()?;
3065        // Collect the set of candidate op_ids referenced by any
3066        // Promote for this sig. Those candidates are no longer
3067        // live.
3068        let mut referenced: std::collections::BTreeSet<lex_vcs::OpId> = Default::default();
3069        for rec in &all {
3070            if let lex_vcs::OperationKind::Promote {
3071                sig_id: s,
3072                winner_candidate,
3073                supersedes,
3074                ..
3075            } = &rec.op.kind
3076            {
3077                if s != sig_id {
3078                    continue;
3079                }
3080                referenced.insert(winner_candidate.clone());
3081                for sup in supersedes {
3082                    referenced.insert(sup.clone());
3083                }
3084            }
3085        }
3086        let mut out: Vec<CandidateInfo> = Vec::new();
3087        for rec in all {
3088            let lex_vcs::OperationKind::Candidate {
3089                sig_id: s,
3090                stage_id,
3091            } = &rec.op.kind
3092            else {
3093                continue;
3094            };
3095            if s != sig_id {
3096                continue;
3097            }
3098            if referenced.contains(&rec.op_id) {
3099                continue;
3100            }
3101            out.push(CandidateInfo {
3102                op_id: rec.op_id.clone(),
3103                stage_id: stage_id.clone(),
3104                intent_id: rec.op.intent_id.clone(),
3105            });
3106        }
3107        out.sort_by(|a, b| a.op_id.cmp(&b.op_id));
3108        Ok(out)
3109    }
3110
3111    /// Promote a previously-landed `Candidate` op as the new
3112    /// branch head for its sig (#294). Emits a `Promote` op
3113    /// listing every other live `Candidate` for the same sig
3114    /// in its `supersedes` field. After this lands,
3115    /// [`Self::list_candidates`] returns an empty set for the
3116    /// sig.
3117    ///
3118    /// Re-typechecks the candidate program (winner stage + the
3119    /// rest of the branch) through `apply_operation_checked`, so
3120    /// a candidate that doesn't compose with the current branch
3121    /// state surfaces as `StoreError::TypeError`.
3122    pub fn promote_candidate(
3123        &self,
3124        branch: &str,
3125        candidate_op_id: &lex_vcs::OpId,
3126    ) -> Result<lex_vcs::OpId, StoreError> {
3127        let log = lex_vcs::OpLog::open(self.root())?;
3128        let candidate_rec = log
3129            .get(candidate_op_id)?
3130            .ok_or_else(|| StoreError::UnknownOp(candidate_op_id.clone()))?;
3131        let (sig, winner_stage_id) = match &candidate_rec.op.kind {
3132            lex_vcs::OperationKind::Candidate { sig_id, stage_id } => {
3133                (sig_id.clone(), stage_id.clone())
3134            }
3135            other => {
3136                return Err(StoreError::InvalidTransition(format!(
3137                    "op `{candidate_op_id}` is a `{:?}`, not a Candidate",
3138                    other
3139                )))
3140            }
3141        };
3142
3143        // #836 G4: a candidate carrying a standing `Reject` review must
3144        // not be promoted. "Standing" = the latest `Review` on the
3145        // winner's stage is a Reject; a later `Approve` (or
3146        // `RequestChanges`, which is advisory, not a veto) lifts it.
3147        // Safe by default: a candidate with no review, or an approved
3148        // one, promotes exactly as before.
3149        if let Some(lex_vcs::ReviewVerdict::Reject) = self.latest_review_verdict(&winner_stage_id)? {
3150            return Err(StoreError::InvalidTransition(format!(
3151                "candidate `{candidate_op_id}` has a standing Reject review on stage                  `{winner_stage_id}`; record an Approve review (or promote a different                  candidate) before promoting"
3152            )));
3153        }
3154
3155        // Gather every OTHER live candidate for this sig — the
3156        // ones this Promote will supersede.
3157        let live = self.list_candidates(&sig)?;
3158        let mut supersedes: Vec<lex_vcs::OpId> = live
3159            .iter()
3160            .filter(|c| &c.op_id != candidate_op_id)
3161            .map(|c| c.op_id.clone())
3162            .collect();
3163        supersedes.sort();
3164
3165        // Assemble candidate program: winner stage in place of
3166        // the sig's current head (if any), plus every other sig
3167        // unchanged.
3168        let head = self.branch_head(branch)?;
3169        let winner_stage = self.get_ast(&winner_stage_id)?;
3170        let mut candidate_program: Vec<lex_ast::Stage> = Vec::with_capacity(head.len() + 1);
3171        let mut found = false;
3172        for (other_sig, other_stage_id) in &head {
3173            if other_sig == &sig {
3174                candidate_program.push(winner_stage.clone());
3175                found = true;
3176            } else {
3177                candidate_program.push(self.get_ast(other_stage_id)?);
3178            }
3179        }
3180        if !found {
3181            // Sig doesn't have a head yet — append the winner
3182            // stage to make it a Create.
3183            candidate_program.push(winner_stage.clone());
3184        }
3185        let from_stage_id = head.get(&sig).cloned();
3186        // Budget delta from old head to winner — same shape as
3187        // ModifyBody.
3188        let from_budget = from_stage_id
3189            .as_deref()
3190            .and_then(|s| self.get_ast(s).ok())
3191            .and_then(|s| budget_of_stage(&s));
3192        let to_budget = budget_of_stage(&winner_stage);
3193
3194        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
3195        let op = lex_vcs::Operation::new(
3196            lex_vcs::OperationKind::Promote {
3197                sig_id: sig.clone(),
3198                winner_candidate: candidate_op_id.clone(),
3199                winner_stage_id: winner_stage_id.clone(),
3200                supersedes,
3201                from_stage_id: from_stage_id.clone(),
3202                from_budget,
3203                to_budget,
3204            },
3205            head_now.into_iter().collect::<Vec<_>>(),
3206        );
3207        let transition = match &from_stage_id {
3208            Some(from) => lex_vcs::StageTransition::Replace {
3209                sig_id: sig,
3210                from: from.clone(),
3211                to: winner_stage_id,
3212            },
3213            None => lex_vcs::StageTransition::Create {
3214                sig_id: sig,
3215                stage_id: winner_stage_id,
3216            },
3217        };
3218        self.apply_operation_checked(branch, op, transition, &candidate_program)
3219    }
3220
3221    /// `set_branch_head_op` for the durability story on the branch
3222    /// file itself.
3223    pub fn apply_operation(
3224        &self,
3225        branch: &str,
3226        op: lex_vcs::Operation,
3227        transition: lex_vcs::StageTransition,
3228    ) -> Result<lex_vcs::OpId, StoreError> {
3229        let attestable = attestable_stage_ids(&transition);
3230        let op_effects = op_declared_effects(&op.kind);
3231        self.cas_retry_advance(branch, op, transition, |new_head| {
3232            self.run_required_attestations_gate(branch, &new_head.op_id, &attestable, &op_effects)
3233        })
3234    }
3235
3236    /// CAS retry loop for #262. Single-parent ops are rebuilt on
3237    /// each iteration with the current branch head as parent;
3238    /// the per-iteration callback runs the gate (and TypeCheck
3239    /// emission, for the checked path) between persist and CAS.
3240    /// Merge ops (with 2 parents already set) skip the rebuild —
3241    /// their parents are caller-supplied and meaningful — and get
3242    /// a single attempt; on CAS failure they surface `Contention`.
3243    fn cas_retry_advance<F>(
3244        &self,
3245        branch: &str,
3246        op: lex_vcs::Operation,
3247        transition: lex_vcs::StageTransition,
3248        mut between_persist_and_cas: F,
3249    ) -> Result<lex_vcs::OpId, StoreError>
3250    where
3251        F: FnMut(&lex_vcs::NewHead) -> Result<(), StoreError>,
3252    {
3253        // 32 retries handles up to ~32 concurrent writers racing on
3254        // the same branch tip. Beyond that, surfacing `Contention`
3255        // is the right signal — clients should back off or batch.
3256        const MAX_ATTEMPTS: u32 = 32;
3257        // Single-parent ops can be rebuilt on retry; merge ops
3258        // can't (their two parents are meaningful, supplied by the
3259        // merge engine). For merges, single attempt: if CAS
3260        // fails, surface Contention.
3261        let is_rebuildable = op.parents.len() <= 1;
3262        let kind = op.kind.clone();
3263        let intent_id = op.intent_id.clone();
3264
3265        let mut last_io_err: Option<StoreError> = None;
3266        let mut current_op = op;
3267        let current_transition = transition;
3268        // Only rebuild on retries — attempt 1 honors the caller's
3269        // exact op so a user-supplied bogus parent (parents =
3270        // ["someone-else"]) surfaces as `StaleParent` instead of
3271        // being silently corrected.
3272        //
3273        // Exception (#262 follow-up): an op with `parents = []`
3274        // means "I don't care; chain off whatever the current
3275        // head is." Under concurrent apply, attempt 1 can read
3276        // `head_op = Some(opA)` after a sibling writer landed,
3277        // and the persist's parent check fails StaleParent
3278        // unprompted. Rebuild attempt 1 for the empty-parents
3279        // case so the legitimate-race path retries cleanly.
3280        let mut rebuilt_already = false;
3281        for attempt in 1..=MAX_ATTEMPTS {
3282            // Read the current head BEFORE we persist — this is
3283            // the value we'll compare against in the CAS.
3284            let parent = self.get_branch(branch)?.and_then(|b| b.head_op);
3285
3286            // Rebuild the op against the current head, but only
3287            // on retries (not the caller's first attempt) and
3288            // only for single-parent operations. Multi-parent
3289            // (merge) ops are passed through unchanged.
3290            //
3291            // Empty-parents ops also rebuild on attempt 1 (see
3292            // the exception note above) so concurrent apply
3293            // doesn't false-positive on StaleParent.
3294            let should_rebuild = is_rebuildable
3295                && (rebuilt_already || (current_op.parents.is_empty() && parent.is_some()));
3296            if should_rebuild {
3297                current_op = lex_vcs::Operation {
3298                    kind: kind.clone(),
3299                    parents: parent.iter().cloned().collect(),
3300                    intent_id: intent_id.clone(),
3301                };
3302            }
3303
3304            // Persist (idempotent). On `StaleParent` from a retry
3305            // attempt (where we already rebuilt), the head changed
3306            // between our `get_branch` and this `lex_vcs::apply`
3307            // — race; rebuild and continue. On `StaleParent` from
3308            // attempt 1 (caller's input), propagate.
3309            let new_head = match self.persist_op_only_with_parent(
3310                branch,
3311                parent.as_ref(),
3312                current_op.clone(),
3313                current_transition.clone(),
3314            ) {
3315                Ok(nh) => nh,
3316                Err(StoreError::Apply(lex_vcs::ApplyError::StaleParent { .. }))
3317                    if is_rebuildable && rebuilt_already =>
3318                {
3319                    rebuilt_already = true;
3320                    continue;
3321                }
3322                Err(e) => return Err(e),
3323            };
3324
3325            // Run the caller's between-persist-and-cas hook
3326            // (TypeCheck emission + gate). If this fails, the op
3327            // record is durable but orphaned — same semantics as
3328            // pre-#262.
3329            between_persist_and_cas(&new_head)?;
3330
3331            // CAS the branch head. On success: done. On mismatch:
3332            // someone advanced in parallel; retry.
3333            match self.set_branch_head_op_cas(branch, parent, new_head.op_id.clone()) {
3334                Ok(()) => return Ok(new_head.op_id),
3335                Err(crate::branches::CasFailed::Mismatch { .. }) if is_rebuildable => {
3336                    // Try again with the new head as parent.
3337                    rebuilt_already = true;
3338                    continue;
3339                }
3340                Err(crate::branches::CasFailed::Mismatch { .. }) => {
3341                    // Merge op: surface immediately — we can't
3342                    // rebuild without rerunning the merge engine.
3343                    let _ = attempt;
3344                    return Err(StoreError::Contention {
3345                        branch: branch.into(),
3346                        attempts: 1,
3347                    });
3348                }
3349                Err(crate::branches::CasFailed::UnknownBranch(b)) => {
3350                    return Err(StoreError::UnknownBranch(b));
3351                }
3352                Err(crate::branches::CasFailed::Io(e)) => {
3353                    last_io_err = Some(StoreError::Io(std::io::Error::other(e)));
3354                    continue;
3355                }
3356            }
3357        }
3358        // Retries exhausted. Prefer surfacing the most recent IO
3359        // error if we hit one; otherwise it's pure CAS contention.
3360        match last_io_err {
3361            Some(e) => Err(e),
3362            None => Err(StoreError::Contention {
3363                branch: branch.into(),
3364                attempts: MAX_ATTEMPTS,
3365            }),
3366        }
3367    }
3368
3369    /// Persist an op against an explicitly-supplied parent. Used
3370    /// by the CAS retry loop in `cas_retry_advance` so the
3371    /// `lex_vcs::apply` parent check matches what we read at the
3372    /// top of the loop iteration (avoids a TOCTOU race against
3373    /// `persist_op_only`'s second read).
3374    fn persist_op_only_with_parent(
3375        &self,
3376        branch: &str,
3377        parent: Option<&lex_vcs::OpId>,
3378        op: lex_vcs::Operation,
3379        transition: lex_vcs::StageTransition,
3380    ) -> Result<lex_vcs::NewHead, StoreError> {
3381        if branch != DEFAULT_BRANCH && self.get_branch(branch)?.is_none() {
3382            return Err(StoreError::UnknownBranch(branch.into()));
3383        }
3384        let log = lex_vcs::OpLog::open(self.root())?;
3385        lex_vcs::apply(&log, parent, op, transition).map_err(|e| match e {
3386            lex_vcs::ApplyError::Persist(io) => StoreError::Io(io),
3387            other => StoreError::Apply(other),
3388        })
3389    }
3390
3391    /// Run the `required_attestations` gate (#245) and the
3392    /// retroactive producer-block gate (#248) over a single op
3393    /// against the store's `policy.json` and attestation log.
3394    ///
3395    /// Failure modes (in order):
3396    ///
3397    /// 1. Producer-block first: if any attestation on the op's
3398    ///    stage is from a quarantined tool, refuse with
3399    ///    `ProducerBlocked` (#248). Surfaces *before* the
3400    ///    required-attestations gate so a clearly-malicious record
3401    ///    isn't masked by a missing-Spec error.
3402    /// 2. Required-attestations next: if any required attestation
3403    ///    kind is missing, refuse with `BranchAdvanceBlocked`
3404    ///    (#245).
3405    ///
3406    /// Loads the policy / attestation log lazily; with no policy
3407    /// file and no `ProducerBlock` attestations the gate is a no-op
3408    /// (default-permissive — matches pre-#245 stores).
3409    fn run_required_attestations_gate(
3410        &self,
3411        branch: &str,
3412        op_id: &lex_vcs::OpId,
3413        stage_ids: &[String],
3414        op_effects: &std::collections::BTreeSet<String>,
3415    ) -> Result<(), StoreError> {
3416        // Build the candidate slice for the new op. Ops with no
3417        // attestable stage (imports, empty merges) get a single
3418        // `None`-stage tuple; both gates skip those.
3419        let new_op_candidate: Vec<(
3420            lex_vcs::OpId,
3421            Option<String>,
3422            std::collections::BTreeSet<String>,
3423        )> = if stage_ids.is_empty() {
3424            vec![(op_id.clone(), None, op_effects.clone())]
3425        } else {
3426            stage_ids
3427                .iter()
3428                .map(|sid| (op_id.clone(), Some(sid.clone()), op_effects.clone()))
3429                .collect()
3430        };
3431        let attest_log = self.attestation_log()?;
3432
3433        // #248 + #256: producer-block gate, walk-back style.
3434        //
3435        // The naive #248 gate only checked the new op's stage. That
3436        // missed contamination on ancestors — once `lex attest
3437        // retro-block` lands, every previously-gated op stays in
3438        // the chain even though its attestations are now from a
3439        // quarantined producer.
3440        //
3441        // #256 fixes this by walking the chain from `head_op` back
3442        // to `last_gate_checkpoint` (or genesis when the checkpoint
3443        // is invalidated), collecting each ancestor's attestable
3444        // stages, and running `check_producer_block` on the
3445        // combined set. After a successful advance,
3446        // `set_branch_head_op` moves the checkpoint to the new
3447        // head (steady-state O(new ops) per advance).
3448        let walk_back_candidate = self.collect_ancestor_candidates(branch)?;
3449        let mut producer_block_candidate = walk_back_candidate;
3450        producer_block_candidate.extend(new_op_candidate.iter().cloned());
3451        crate::policy::check_producer_block(&attest_log, &producer_block_candidate)
3452            .map_err(StoreError::ProducerBlocked)?;
3453
3454        // #245: required-attestations gate. Forward-going only —
3455        // only the new op is checked. Walking back makes no sense
3456        // here: the policy is "this advance must carry these
3457        // attestations," not "every prior op must have."
3458        let policy = match crate::policy::load(self.root())? {
3459            Some(p) if !p.required_attestations.is_empty() => p,
3460            _ => return Ok(()),
3461        };
3462        let waivers =
3463            crate::policy::check_required_attestations(&attest_log, &new_op_candidate, &policy)
3464                .map_err(StoreError::BranchAdvanceBlocked)?;
3465        // #293: emit one `TrustWaived` attestation per waiver so
3466        // the audit trail records every skip. Idempotent on
3467        // attestation_id (content-addressed dedup) — re-running
3468        // the gate with the same state writes the same files.
3469        for w in waivers {
3470            let att = lex_vcs::Attestation::new(
3471                w.stage_id,
3472                Some(op_id.clone()),
3473                None,
3474                lex_vcs::AttestationKind::TrustWaived {
3475                    producer: w.producer,
3476                    score_thousandths: w.score_thousandths,
3477                    threshold_thousandths: w.threshold_thousandths,
3478                    kind_tag: w.kind_tag,
3479                },
3480                lex_vcs::AttestationResult::Passed,
3481                trust_waived_producer(),
3482                None,
3483            );
3484            attest_log.put(&att)?;
3485        }
3486        Ok(())
3487    }
3488
3489    /// Walk the branch from `head_op` back to `last_gate_checkpoint`
3490    /// (exclusive) and return the `(op_id, stage_id, op_effects)`
3491    /// tuples for every attestable stage touched by an ancestor
3492    /// (#256). Empty when the branch is fresh, when the checkpoint
3493    /// equals the head, or when the head is None.
3494    fn collect_ancestor_candidates(&self, branch: &str) -> Result<Vec<GateCandidate>, StoreError> {
3495        let b = match self.get_branch(branch)? {
3496            Some(b) => b,
3497            None => return Ok(Vec::new()),
3498        };
3499        let Some(head) = b.head_op else {
3500            return Ok(Vec::new());
3501        };
3502        if Some(&head) == b.last_gate_checkpoint.as_ref() {
3503            // Steady-state common case: previous advance left the
3504            // checkpoint at head. Nothing to re-walk.
3505            return Ok(Vec::new());
3506        }
3507
3508        let log = lex_vcs::OpLog::open(self.root())?;
3509        let walk = log.walk_back(&head, None)?;
3510        let stop_at = b.last_gate_checkpoint.clone();
3511        let mut out = Vec::new();
3512        for rec in walk {
3513            if Some(&rec.op_id) == stop_at.as_ref() {
3514                break;
3515            }
3516            let stages = attestable_stage_ids(&rec.produces);
3517            let effects = op_declared_effects(&rec.op.kind);
3518            if stages.is_empty() {
3519                out.push((rec.op_id.clone(), None, effects));
3520            } else {
3521                for sid in stages {
3522                    out.push((rec.op_id.clone(), Some(sid), effects.clone()));
3523                }
3524            }
3525        }
3526        Ok(out)
3527    }
3528}
3529
3530fn stage_name(stage: &Stage) -> &str {
3531    match stage {
3532        Stage::FnDecl(fd) => &fd.name,
3533        Stage::TypeDecl(td) => &td.name,
3534        Stage::Import(i) => &i.alias,
3535    }
3536}
3537
3538fn stage_for_kind<'a>(
3539    kind: &lex_vcs::OperationKind,
3540    stages: &'a [lex_ast::Stage],
3541) -> Option<&'a lex_ast::Stage> {
3542    use lex_vcs::OperationKind::*;
3543    let target_sig = match kind {
3544        AddFunction { sig_id, .. }
3545        | ModifyBody { sig_id, .. }
3546        | ChangeEffectSig { sig_id, .. }
3547        | AddType { sig_id, .. }
3548        | ModifyType { sig_id, .. } => Some(sig_id.clone()),
3549        RenameSymbol { to, .. } => Some(to.clone()),
3550        _ => None,
3551    };
3552    let target_sig = target_sig?;
3553    stages
3554        .iter()
3555        .find(|s| sig_id(s).as_deref() == Some(target_sig.as_str()))
3556}
3557
3558fn transition_for_kind(kind: &lex_vcs::OperationKind) -> lex_vcs::StageTransition {
3559    use lex_vcs::OperationKind::*;
3560    use lex_vcs::StageTransition;
3561    match kind {
3562        AddFunction {
3563            sig_id, stage_id, ..
3564        }
3565        | AddType { sig_id, stage_id, .. } => StageTransition::Create {
3566            sig_id: sig_id.clone(),
3567            stage_id: stage_id.clone(),
3568        },
3569        RemoveFunction {
3570            sig_id,
3571            last_stage_id,
3572        }
3573        | RemoveType {
3574            sig_id,
3575            last_stage_id,
3576        } => StageTransition::Remove {
3577            sig_id: sig_id.clone(),
3578            last: last_stage_id.clone(),
3579        },
3580        ModifyBody {
3581            sig_id,
3582            from_stage_id,
3583            to_stage_id,
3584            ..
3585        }
3586        | ChangeEffectSig {
3587            sig_id,
3588            from_stage_id,
3589            to_stage_id,
3590            ..
3591        }
3592        | ModifyType {
3593            sig_id,
3594            from_stage_id,
3595            to_stage_id,
3596        }
3597        | ReplaceMatchArm {
3598            sig_id,
3599            from_stage_id,
3600            to_stage_id,
3601            ..
3602        }
3603        | RenameLocal {
3604            sig_id,
3605            from_stage_id,
3606            to_stage_id,
3607            ..
3608        }
3609        | InlineLet {
3610            sig_id,
3611            from_stage_id,
3612            to_stage_id,
3613            ..
3614        } => StageTransition::Replace {
3615            sig_id: sig_id.clone(),
3616            from: from_stage_id.clone(),
3617            to: to_stage_id.clone(),
3618        },
3619        RenameSymbol {
3620            from,
3621            to,
3622            body_stage_id,
3623        } => StageTransition::Rename {
3624            from: from.clone(),
3625            to: to.clone(),
3626            body_stage_id: body_stage_id.clone(),
3627        },
3628        AddImport { .. } | RemoveImport { .. } => StageTransition::ImportOnly,
3629        Merge { .. } => StageTransition::Merge {
3630            entries: Default::default(),
3631        },
3632        // #294: a Candidate proposes a stage without advancing
3633        // the branch. ImportOnly keeps the branch head untouched
3634        // — the stage IS published on disk (Store::propose_candidate
3635        // calls publish before apply), but no head delta lands.
3636        Candidate { .. } => StageTransition::ImportOnly,
3637        // A Promote advances the head exactly like ModifyBody
3638        // (or Create when the sig had no head). The winner
3639        // stage is the new branch state for that sig.
3640        Promote {
3641            sig_id,
3642            winner_stage_id,
3643            from_stage_id,
3644            ..
3645        } => match from_stage_id {
3646            Some(from) => StageTransition::Replace {
3647                sig_id: sig_id.clone(),
3648                from: from.clone(),
3649                to: winner_stage_id.clone(),
3650            },
3651            None => StageTransition::Create {
3652                sig_id: sig_id.clone(),
3653                stage_id: winner_stage_id.clone(),
3654            },
3655        },
3656    }
3657}
3658
3659/// Producer identity for TypeCheck attestations emitted by the
3660/// store-write gate. Pinned to this crate's name + version so an
3661/// attestation produced by a different `lex-store` revision is
3662/// distinguishable (content-hashed `produced_by`).
3663fn typecheck_producer() -> lex_vcs::ProducerDescriptor {
3664    lex_vcs::ProducerDescriptor {
3665        tool: "lex-store".into(),
3666        version: env!("CARGO_PKG_VERSION").into(),
3667        model: None,
3668    }
3669}
3670
3671/// Producer for attestations the hosted CI runner writes (#93). A
3672/// distinct tool name so a `require-attestation` gate — via the
3673/// producer-trust model — can weight "the hub verified this
3674/// server-side" above a client-attached `TypeCheck`.
3675fn hub_ci_producer() -> lex_vcs::ProducerDescriptor {
3676    lex_vcs::ProducerDescriptor {
3677        tool: "lex-hub-ci".into(),
3678        version: env!("CARGO_PKG_VERSION").into(),
3679        model: None,
3680    }
3681}
3682
3683/// Verdict of a hosted-CI run over a branch head (#93).
3684#[derive(Debug, Clone, serde::Serialize)]
3685pub struct HubCiVerdict {
3686    pub passed: bool,
3687    pub checked_stages: usize,
3688    pub attested_stages: usize,
3689    #[serde(skip_serializing_if = "Option::is_none")]
3690    pub detail: Option<String>,
3691}
3692
3693/// Producer for the replay-comparison attestation (#836 G3). Distinct
3694/// tool name so the comparison lex performed is attributable
3695/// separately from the (external) regeneration.
3696fn replay_producer() -> lex_vcs::ProducerDescriptor {
3697    lex_vcs::ProducerDescriptor {
3698        tool: "lex-store-replay".into(),
3699        version: env!("CARGO_PKG_VERSION").into(),
3700        model: None,
3701    }
3702}
3703
3704/// Human/audit label for a recorded model: `provider/name` (`@version`
3705/// when pinned).
3706fn model_label(m: &lex_vcs::ModelDescriptor) -> String {
3707    match &m.version {
3708        Some(v) => format!("{}/{}@{}", m.provider, m.name, v),
3709        None => format!("{}/{}", m.provider, m.name),
3710    }
3711}
3712
3713/// The `(sig_id, stage_id)` an op recorded producing, or `None` for a
3714/// transition that produces no stage (removal / import / merge) — those
3715/// have nothing to regenerate for a replay.
3716fn produced_sig_stage(t: &lex_vcs::StageTransition) -> Option<(String, String)> {
3717    use lex_vcs::StageTransition::*;
3718    match t {
3719        Create { sig_id, stage_id } => Some((sig_id.clone(), stage_id.clone())),
3720        Replace { sig_id, to, .. } => Some((sig_id.clone(), to.clone())),
3721        Rename { to, body_stage_id, .. } => Some((to.clone(), body_stage_id.clone())),
3722        Remove { .. } | ImportOnly | Merge { .. } => None,
3723    }
3724}
3725
3726/// Producer identity for `Examples::Passed` attestations emitted by
3727/// [`Store::record_examples_passed`] (#835). Distinct tool name so
3728/// the activity feed can tell an auto-emitted publish-time examples
3729/// verdict apart from an `lex agent-tool --examples` one.
3730fn examples_producer() -> lex_vcs::ProducerDescriptor {
3731    lex_vcs::ProducerDescriptor {
3732        tool: "lex-store::examples".into(),
3733        version: env!("CARGO_PKG_VERSION").into(),
3734        model: None,
3735    }
3736}
3737
3738/// Producer identity for `Review` attestations (#836). The reviewer's
3739/// own id lives in the kind; this records which tool minted the record.
3740fn review_producer(reviewer: &str) -> lex_vcs::ProducerDescriptor {
3741    lex_vcs::ProducerDescriptor {
3742        tool: format!("lex-store::review:{reviewer}"),
3743        version: env!("CARGO_PKG_VERSION").into(),
3744        model: None,
3745    }
3746}
3747
3748/// Producer identity for `RepairHint` attestations emitted by
3749/// `apply_operation_checked` on TypeError (#281). Distinct tool
3750/// name from `typecheck_producer` so consumers can filter the
3751/// activity feed for repair hints without scanning kinds.
3752fn repair_hint_producer() -> lex_vcs::ProducerDescriptor {
3753    lex_vcs::ProducerDescriptor {
3754        tool: "lex-store::repair_hint".into(),
3755        version: env!("CARGO_PKG_VERSION").into(),
3756        model: None,
3757    }
3758}
3759
3760/// Producer identity for `TrustWaived` attestations emitted by
3761/// the `required_attestations` gate on a trust-driven waiver
3762/// (#293). Distinct from `typecheck_producer` and `repair_hint`
3763/// so the audit trail clearly shows "the gate let this advance
3764/// through because trust > threshold."
3765fn trust_waived_producer() -> lex_vcs::ProducerDescriptor {
3766    lex_vcs::ProducerDescriptor {
3767        tool: "lex-store::trust_waived".into(),
3768        version: env!("CARGO_PKG_VERSION").into(),
3769        model: None,
3770    }
3771}
3772
3773/// Producer identity for `ProducerTrust` attestations emitted by
3774/// [`Store::recompute_producer_trust`]. The score-derivation
3775/// recompute is its own machine-emittable kind, distinct from
3776/// the gate-side `TrustWaived` emit (#293).
3777fn producer_trust_producer() -> lex_vcs::ProducerDescriptor {
3778    lex_vcs::ProducerDescriptor {
3779        tool: "lex-store::producer_trust".into(),
3780        version: env!("CARGO_PKG_VERSION").into(),
3781        model: None,
3782    }
3783}
3784
3785/// The set of stage_ids a transition introduces. These are the
3786/// stages a successful TypeCheck pass attests *about* — the new
3787/// head produced by Create/Replace, the renamed body, or the per-
3788/// sig resolution of a Merge. Removes and ImportOnly produce no
3789/// attestable stage; the program typechecks but no specific stage
3790/// is the subject of the claim.
3791/// One row of input to the producer-block / required-attestations
3792/// gates: `(op_id, stage_id, op_effects)`. The `stage_id` is
3793/// `None` for ops that don't touch a stage (imports, empty
3794/// merges) — the gate skips those.
3795type GateCandidate = (
3796    lex_vcs::OpId,
3797    Option<String>,
3798    std::collections::BTreeSet<String>,
3799);
3800
3801/// Effect set declared *by the operation itself* (#245). Used by
3802/// the `required_attestations` gate's `EffectsIntersect` clause.
3803///
3804/// Only `AddFunction` and `ChangeEffectSig` carry an effect set in
3805/// their op payload; for everything else this returns the empty
3806/// set, which means `EffectsIntersect` rules don't fire on those
3807/// ops. `Always` rules continue to fire regardless. A future
3808/// improvement is to extract effects from the candidate `Stage`
3809/// for `ModifyBody` ops, but the typed-effects-on-ops path (#247)
3810/// is the cleaner solution and lands separately.
3811fn op_declared_effects(kind: &lex_vcs::OperationKind) -> std::collections::BTreeSet<String> {
3812    use lex_vcs::OperationKind::*;
3813    match kind {
3814        AddFunction { effects, .. } => effects.clone(),
3815        ChangeEffectSig { to_effects, .. } => to_effects.clone(),
3816        _ => std::collections::BTreeSet::new(),
3817    }
3818}
3819
3820fn attestable_stage_ids(transition: &lex_vcs::StageTransition) -> Vec<String> {
3821    use lex_vcs::StageTransition::*;
3822    match transition {
3823        Create { stage_id, .. } => vec![stage_id.clone()],
3824        Replace { to, .. } => vec![to.clone()],
3825        Rename { body_stage_id, .. } => vec![body_stage_id.clone()],
3826        Merge { entries } => entries.values().filter_map(|opt| opt.clone()).collect(),
3827        Remove { .. } | ImportOnly => Vec::new(),
3828    }
3829}
3830
3831/// True when two `FnDecl`s are identical except for their body — the
3832/// precondition for a pure intra-body three-way merge (#838). The
3833/// signature fields are equal by construction when both share a
3834/// `sig_id`; this also guards the non-signature fields (`type_params`,
3835/// `examples`) so a side that changed those isn't silently dropped.
3836fn fndecl_same_except_body(a: &lex_ast::FnDecl, b: &lex_ast::FnDecl) -> bool {
3837    a.name == b.name
3838        && a.type_params == b.type_params
3839        && a.params == b.params
3840        && a.effects == b.effects
3841        && a.effect_row_var == b.effect_row_var
3842        && a.return_type == b.return_type
3843        && a.examples == b.examples
3844}
3845
3846fn write_canonical_json<T: Serialize>(path: &Path, value: &T) -> Result<(), StoreError> {
3847    let v = serde_json::to_value(value)?;
3848    let s = lex_ast::canon_json::to_canonical_string(&v);
3849    if let Some(parent) = path.parent() {
3850        fs::create_dir_all(parent)?;
3851    }
3852    fs::write(path, s)?;
3853    Ok(())
3854}
3855
3856/// Read the `name` of the `Let` expression at `let_node` inside
3857/// `stage`'s body. Used by [`Store::apply_rename_local`] to record
3858/// the rename source. Returns the same `TransformError` shapes as
3859/// the transformer itself so callers see a consistent error
3860/// vocabulary.
3861fn read_let_name(
3862    stage: &Stage,
3863    let_node: &lex_ast::NodeId,
3864) -> Result<String, lex_ast::TransformError> {
3865    // The transformer is itself a pure function; ask it to perform
3866    // a rename to a sentinel value and read the resulting let's
3867    // original name from the output. Cheaper than duplicating the
3868    // node-walk here, and stays correct as the transform evolves.
3869    //
3870    // We use a sentinel that's invalid as a Lex identifier so even
3871    // if the rename somehow lands, downstream parsing would
3872    // surface it loudly. (The transform path discards the renamed
3873    // value — we only need the *original* name.)
3874    let probed = lex_ast::rename_local(stage, let_node, "__lex_rename_probe__")?;
3875    let Stage::FnDecl(fd) = probed else {
3876        return Err(lex_ast::TransformError::NonFnTarget {
3877            stage_kind: "non-FnDecl",
3878        });
3879    };
3880    // Walk back to the probed let to read its old name from the
3881    // *original* stage — the probed stage's let has already been
3882    // renamed.
3883    let Stage::FnDecl(orig_fd) = stage else {
3884        return Err(lex_ast::TransformError::NonFnTarget {
3885            stage_kind: "non-FnDecl",
3886        });
3887    };
3888    // Path-based lookup matches the transformer's navigation.
3889    let path = parse_let_node_path(let_node.as_str())?;
3890    if path.is_empty() {
3891        return Err(lex_ast::TransformError::NotALet {
3892            at: let_node.as_str().into(),
3893            found_kind: "stage_root",
3894        });
3895    }
3896    if path[0] != orig_fd.params.len() + 1 {
3897        return Err(lex_ast::TransformError::UnknownNode {
3898            at: let_node.as_str().into(),
3899        });
3900    }
3901    let inner = &path[1..];
3902    let target = navigate_to_let(&orig_fd.body, inner, let_node.as_str())?;
3903    let _ = fd; // probed stage discarded
3904    Ok(target.to_string())
3905}
3906
3907fn parse_let_node_path(id: &str) -> Result<Vec<usize>, lex_ast::TransformError> {
3908    let s = id
3909        .strip_prefix("n_")
3910        .ok_or_else(|| lex_ast::TransformError::BadNodeId(id.into()))?;
3911    let mut parts = s.split('.');
3912    let head = parts
3913        .next()
3914        .ok_or_else(|| lex_ast::TransformError::BadNodeId(id.into()))?;
3915    if head != "0" {
3916        return Err(lex_ast::TransformError::BadNodeId(id.into()));
3917    }
3918    let mut out = Vec::new();
3919    for p in parts {
3920        out.push(
3921            p.parse::<usize>()
3922                .map_err(|_| lex_ast::TransformError::BadNodeId(id.into()))?,
3923        );
3924    }
3925    Ok(out)
3926}
3927
3928fn navigate_to_let<'a>(
3929    root: &'a lex_ast::CExpr,
3930    path: &[usize],
3931    at: &str,
3932) -> Result<&'a str, lex_ast::TransformError> {
3933    use lex_ast::CExpr::*;
3934    let mut current = root;
3935    for &idx in path {
3936        current = match current {
3937            Call { callee, args } => {
3938                if idx == 0 {
3939                    callee
3940                } else {
3941                    args.get(idx - 1)
3942                        .ok_or_else(|| lex_ast::TransformError::UnknownNode { at: at.into() })?
3943                }
3944            }
3945            Let { value, body, .. } => match idx {
3946                0 => value,
3947                1 => body,
3948                _ => return Err(lex_ast::TransformError::UnknownNode { at: at.into() }),
3949            },
3950            Match { scrutinee, arms } => {
3951                if idx == 0 {
3952                    scrutinee
3953                } else {
3954                    let arm_off = idx - 1;
3955                    if arm_off % 2 != 1 {
3956                        return Err(lex_ast::TransformError::UnknownNode { at: at.into() });
3957                    }
3958                    let arm_index = arm_off / 2;
3959                    &arms
3960                        .get(arm_index)
3961                        .ok_or_else(|| lex_ast::TransformError::UnknownNode { at: at.into() })?
3962                        .body
3963                }
3964            }
3965            Block { statements, result } => {
3966                if idx < statements.len() {
3967                    &statements[idx]
3968                } else if idx == statements.len() {
3969                    result
3970                } else {
3971                    return Err(lex_ast::TransformError::UnknownNode { at: at.into() });
3972                }
3973            }
3974            Constructor { args, .. }
3975            | TupleLit { items: args, .. }
3976            | ListLit { items: args, .. } => args
3977                .get(idx)
3978                .ok_or_else(|| lex_ast::TransformError::UnknownNode { at: at.into() })?,
3979            RecordLit { fields } => {
3980                &fields
3981                    .get(idx)
3982                    .ok_or_else(|| lex_ast::TransformError::UnknownNode { at: at.into() })?
3983                    .value
3984            }
3985            FieldAccess { value, .. } if idx == 0 => value,
3986            Lambda { body, .. } if idx == 0 => body,
3987            BinOp { lhs, rhs, .. } => match idx {
3988                0 => lhs,
3989                1 => rhs,
3990                _ => return Err(lex_ast::TransformError::UnknownNode { at: at.into() }),
3991            },
3992            UnaryOp { expr, .. } if idx == 0 => expr,
3993            Return { value } if idx == 0 => value,
3994            _ => return Err(lex_ast::TransformError::UnknownNode { at: at.into() }),
3995        };
3996    }
3997    let Let { name, .. } = current else {
3998        return Err(lex_ast::TransformError::NotALet {
3999            at: at.into(),
4000            found_kind: lex_cexpr_kind(current),
4001        });
4002    };
4003    Ok(name)
4004}
4005
4006fn lex_cexpr_kind(e: &lex_ast::CExpr) -> &'static str {
4007    use lex_ast::CExpr::*;
4008    match e {
4009        Literal { .. } => "Literal",
4010        Var { .. } => "Var",
4011        Call { .. } => "Call",
4012        Let { .. } => "Let",
4013        Match { .. } => "Match",
4014        Block { .. } => "Block",
4015        Constructor { .. } => "Constructor",
4016        RecordLit { .. } => "RecordLit",
4017        TupleLit { .. } => "TupleLit",
4018        ListLit { .. } => "ListLit",
4019        FieldAccess { .. } => "FieldAccess",
4020        Lambda { .. } => "Lambda",
4021        BinOp { .. } => "BinOp",
4022        UnaryOp { .. } => "UnaryOp",
4023        Return { .. } => "Return",
4024    }
4025}
4026
4027/// Extract the declared `[budget(N)]` integer from a stage's
4028/// effect set, if any (#280 + #247). Returns `None` for stages
4029/// that aren't `FnDecl` or don't carry a budget effect — same
4030/// shape as `lex_vcs::budget_from_effects`.
4031fn budget_of_stage(stage: &Stage) -> Option<u64> {
4032    let fd = match stage {
4033        Stage::FnDecl(fd) => fd,
4034        _ => return None,
4035    };
4036    let mut min_cost: Option<u64> = None;
4037    for eff in &fd.effects {
4038        if eff.name != "budget" {
4039            continue;
4040        }
4041        if let Some(lex_ast::EffectArg::Int { value }) = &eff.arg {
4042            let n = *value as u64;
4043            min_cost = Some(min_cost.map(|c| c.min(n)).unwrap_or(n));
4044        }
4045    }
4046    min_cost
4047}
4048
4049/// Serialize a stage to its canonical-JSON byte form. Used by
4050/// `publish_signed` for delta encoding (#261 slice 3) — both the
4051/// "compute the diff" path and the "write a full snapshot"
4052/// fallback need exactly the same bytes.
4053fn canonical_bytes(stage: &Stage) -> Result<Vec<u8>, StoreError> {
4054    let v = serde_json::to_value(stage)?;
4055    Ok(lex_ast::canon_json::to_canonical_string(&v).into_bytes())
4056}
4057
4058#[allow(dead_code)]
4059fn read_json<T: DeserializeOwned>(path: &Path) -> Result<T, StoreError> {
4060    let bytes = fs::read(path)?;
4061    Ok(serde_json::from_slice(&bytes)?)
4062}