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