Skip to main content

lex_store/
store.rs

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