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