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::fs;
15use std::path::{Path, PathBuf};
16use std::time::{SystemTime, UNIX_EPOCH};
17
18#[derive(Debug, thiserror::Error)]
19pub enum StoreError {
20    #[error("io error: {0}")]
21    Io(#[from] std::io::Error),
22    #[error("serialization error: {0}")]
23    Serde(#[from] serde_json::Error),
24    #[error("imports cannot be published as stages")]
25    CannotPublishImport,
26    #[error("unknown stage_id `{0}`")]
27    UnknownStage(String),
28    #[error("unknown sig_id `{0}`")]
29    UnknownSig(String),
30    #[error("invalid lifecycle transition: {0}")]
31    InvalidTransition(String),
32    #[error("unknown branch `{0}`")]
33    UnknownBranch(String),
34    #[error("unknown blob `{0}`")]
35    UnknownBlob(String),
36    #[error("unknown blob ref `{namespace}/{key}`")]
37    UnknownBlobRef { namespace: String, key: String },
38    #[error("unknown op_id `{0}`")]
39    UnknownOp(lex_vcs::OpId),
40    /// A typed AST transform (#280) — e.g. `ReplaceMatchArm` — was
41    /// asked to operate on a node it couldn't address (wrong kind,
42    /// out-of-range arm index, unknown NodeId, etc.). Distinct from
43    /// `TypeError` (which means the transform succeeded but its
44    /// output didn't typecheck) so callers can render the right
45    /// error message.
46    #[error("transform failed: {0}")]
47    TransformError(lex_ast::TransformError),
48    #[error(transparent)]
49    Apply(#[from] lex_vcs::ApplyError),
50    /// The candidate program — i.e. the source the caller is
51    /// publishing — doesn't typecheck. The branch head is unchanged
52    /// and no op records are persisted. Issue #130's "always-valid
53    /// HEAD" invariant: the gate runs before any side effect, so a
54    /// type-broken publish leaves no footprint.
55    #[error("type errors in published program: {} error(s)", .0.len())]
56    TypeError(Vec<lex_types::TypeError>),
57    /// The op was persisted but a `required_attestations` rule in
58    /// `policy.json` (#245) refused to advance the branch head past
59    /// it. The op record is durable — re-running with the missing
60    /// attestations recorded will succeed without re-persisting —
61    /// but the branch is unchanged. Surfaced as a structured JSON
62    /// envelope on the HTTP API.
63    #[error(
64        "branch advance blocked: op {} missing attestations: {}",
65        .0.op_id, .0.missing.join(", ")
66    )]
67    BranchAdvanceBlocked(crate::policy::BranchAdvanceBlocked),
68    /// All retry attempts of the CAS branch-head advance failed
69    /// because another writer kept advancing the same branch
70    /// (#262). The op record itself is durable in the op log
71    /// (orphaned), so re-running with backoff would eventually
72    /// land — return `503 Contention { retry_after }` from the
73    /// HTTP API and let the client back off.
74    #[error("branch advance contention on `{branch}`: {attempts} retries exhausted")]
75    Contention { branch: String, attempts: u32 },
76    /// The op was persisted but its stage carries an attestation
77    /// produced by a retroactively quarantined tool (#248). The
78    /// branch head is unchanged. The op record stays in the log
79    /// (audit trail intact); re-running with the producer
80    /// unblocked, or with un-contaminated attestations, succeeds
81    /// without re-persisting the op.
82    #[error(
83        "branch advance blocked: op {} touches stage {} with an attestation from \
84         quarantined producer `{}` (blocked at {}, attestation at {})",
85        .0.op_id, .0.stage_id, .0.tool_id, .0.blocked_at, .0.attestation_at
86    )]
87    ProducerBlocked(crate::policy::ProducerBlocked),
88    /// The op would push its session's monotonic budget over the
89    /// cap configured in `policy.session_budgets` (#292 slice 3).
90    /// The op is *not* persisted; the branch head is unchanged.
91    /// The caller should either start a new session, raise the
92    /// cap, or refactor to fit the budget. HTTP API maps to 503.
93    #[error("session `{session_id}` budget exceeded: spent_after={spent_after} > cap={cap}")]
94    BudgetExceeded {
95        session_id: String,
96        cap: u64,
97        spent_after: u64,
98    },
99}
100
101/// The outcome returned by [`Store::publish_program`].
102#[derive(Debug, Clone, serde::Serialize)]
103pub struct PublishOutcome {
104    pub ops: Vec<PublishOp>,
105    pub head_op: Option<lex_vcs::OpId>,
106}
107
108/// One applied operation within a [`PublishOutcome`].
109#[derive(Debug, Clone, serde::Serialize)]
110pub struct PublishOp {
111    pub op_id: lex_vcs::OpId,
112    pub kind: serde_json::Value,
113}
114
115/// One entry in the per-`SigId` stage history surfaced by
116/// `Store::sig_history`. Newest-first ordering is the responsibility
117/// of the producer.
118#[derive(Debug, Clone, serde::Serialize, PartialEq)]
119pub struct StageHistoryEntry {
120    pub stage_id: String,
121    pub status: StageStatus,
122    /// Wall-clock seconds of the most recent transition.
123    pub last_at: u64,
124    /// Wall-clock seconds when this stage was first written to the
125    /// store (its initial Draft transition). `None` for stages
126    /// whose lifecycle log doesn't include an explicit Draft entry
127    /// — shouldn't happen for stages published via `Store::publish`,
128    /// but the type allows hand-edited stores.
129    #[serde(skip_serializing_if = "Option::is_none")]
130    pub published_at: Option<u64>,
131}
132
133/// Per-candidate metadata surfaced by [`Store::list_candidates`]
134/// (#294). Returned sorted by `op_id` for deterministic output.
135#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
136pub struct CandidateInfo {
137    pub op_id: lex_vcs::OpId,
138    pub stage_id: lex_vcs::StageId,
139    /// Author intent. Always set for `Candidate` ops emitted via
140    /// [`Store::propose_candidate`]; `None` only if a
141    /// hand-written raw op skipped the intent tag.
142    pub intent_id: Option<lex_vcs::IntentId>,
143}
144
145pub struct Store {
146    root: PathBuf,
147}
148
149impl Store {
150    /// Open or create a store rooted at `root`.
151    pub fn open(root: impl AsRef<Path>) -> Result<Self, StoreError> {
152        let root = root.as_ref().to_path_buf();
153        fs::create_dir_all(root.join("stages"))?;
154        fs::create_dir_all(root.join("traces"))?;
155        Ok(Self { root })
156    }
157
158    pub fn root(&self) -> &Path {
159        &self.root
160    }
161
162    // ── Generic content-addressed blobs (#5 / M6.1) ──────────────────────────
163    //
164    // The stage store holds typed Lex ASTs; loom-style artifacts (generated
165    // code, JSON, prose) are opaque text. These blob methods give the store a
166    // generic content-addressed object alongside stages, plus a lightweight
167    // ref namespace so callers can bind names (e.g. a sprint's node ids) to
168    // blob shas without touching the operation-log branch machinery.
169    //
170    // The sha is the lowercase hex SHA-256 of the content's UTF-8 bytes —
171    // identical to Lex's `crypto.sha256_str`, so a blob written here and an
172    // artifact content-addressed in loom's SQLite store share the same id and
173    // are interchangeable by reference. Store-scoped, so under lex-hub each
174    // tenant store gets its own blob space for free.
175
176    fn blobs_dir(&self) -> PathBuf {
177        self.root.join("blobs")
178    }
179
180    fn blob_refs_dir(&self) -> PathBuf {
181        self.root.join("blobrefs")
182    }
183
184    /// Content-address `content` and persist it under `<root>/blobs/<sha>`.
185    /// Returns the sha. Idempotent: re-putting identical content is a no-op.
186    /// Concurrency-safe — writes to a unique temp file then atomically renames
187    /// onto the content-addressed path, so parallel writers of the same content
188    /// can't corrupt it.
189    pub fn put_blob(&self, content: &str) -> Result<String, StoreError> {
190        use sha2::{Digest, Sha256};
191        let sha = hex::encode(Sha256::digest(content.as_bytes()));
192        let dir = self.blobs_dir();
193        let path = dir.join(&sha);
194        if path.exists() {
195            return Ok(sha);
196        }
197        fs::create_dir_all(&dir)?;
198        static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
199        let n = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
200        let tmp = dir.join(format!(".{sha}.{}.{n}.tmp", std::process::id()));
201        fs::write(&tmp, content.as_bytes())?;
202        // rename is atomic on the same filesystem; identical content makes a
203        // last-writer-wins race harmless.
204        fs::rename(&tmp, &path)?;
205        Ok(sha)
206    }
207
208    /// Read a blob by its sha. `UnknownBlob` if absent.
209    pub fn get_blob(&self, sha: &str) -> Result<String, StoreError> {
210        match fs::read_to_string(self.blobs_dir().join(sha)) {
211            Ok(s) => Ok(s),
212            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
213                Err(StoreError::UnknownBlob(sha.to_string()))
214            }
215            Err(e) => Err(StoreError::Io(e)),
216        }
217    }
218
219    /// Whether a blob with this sha exists.
220    pub fn has_blob(&self, sha: &str) -> bool {
221        self.blobs_dir().join(sha).exists()
222    }
223
224    /// Bind `key` to a blob `sha` within `namespace` (e.g. namespace
225    /// `"loom/sprint-abc"`, key `"build-node"`). Overwrites an existing
226    /// binding. The namespace may contain `/`; neither namespace nor key may
227    /// contain a `..` path component.
228    pub fn set_blob_ref(&self, namespace: &str, key: &str, sha: &str) -> Result<(), StoreError> {
229        let dir = self.blob_ref_namespace_dir(namespace, key)?;
230        fs::create_dir_all(&dir)?;
231        fs::write(dir.join(key), sha.as_bytes())?;
232        Ok(())
233    }
234
235    /// Resolve `namespace`/`key` to a blob sha. `UnknownBlobRef` if unbound.
236    pub fn get_blob_ref(&self, namespace: &str, key: &str) -> Result<String, StoreError> {
237        let dir = self.blob_ref_namespace_dir(namespace, key)?;
238        match fs::read_to_string(dir.join(key)) {
239            Ok(s) => Ok(s.trim().to_string()),
240            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Err(StoreError::UnknownBlobRef {
241                namespace: namespace.to_string(),
242                key: key.to_string(),
243            }),
244            Err(e) => Err(StoreError::Io(e)),
245        }
246    }
247
248    /// All `key → sha` bindings in a namespace (e.g. every artifact in a
249    /// sprint). Empty map if the namespace has no bindings yet.
250    pub fn list_blob_refs(
251        &self,
252        namespace: &str,
253    ) -> Result<std::collections::BTreeMap<String, String>, StoreError> {
254        let dir = self.blob_ref_namespace_dir(namespace, "x")?;
255        let mut out = std::collections::BTreeMap::new();
256        let entries = match fs::read_dir(&dir) {
257            Ok(e) => e,
258            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out),
259            Err(e) => return Err(StoreError::Io(e)),
260        };
261        for entry in entries {
262            let entry = entry?;
263            if entry.file_type()?.is_file() {
264                let key = entry.file_name().to_string_lossy().to_string();
265                let sha = fs::read_to_string(entry.path())?.trim().to_string();
266                out.insert(key, sha);
267            }
268        }
269        Ok(out)
270    }
271
272    // Resolve the on-disk dir for a (namespace, key), rejecting `..` traversal
273    // and `/` in the key. `key` is validated but not joined here (callers join
274    // it themselves so `list_blob_refs` can pass a dummy).
275    fn blob_ref_namespace_dir(&self, namespace: &str, key: &str) -> Result<PathBuf, StoreError> {
276        if key.contains('/') || key.contains('\\') || key.split('/').any(|c| c == "..") {
277            return Err(StoreError::UnknownBlobRef {
278                namespace: namespace.to_string(),
279                key: key.to_string(),
280            });
281        }
282        let mut dir = self.blob_refs_dir();
283        for comp in namespace.split('/') {
284            if comp == ".." || comp.contains('\\') {
285                return Err(StoreError::UnknownBlobRef {
286                    namespace: namespace.to_string(),
287                    key: key.to_string(),
288                });
289            }
290            if !comp.is_empty() {
291                dir.push(comp);
292            }
293        }
294        Ok(dir)
295    }
296
297    fn now() -> u64 {
298        SystemTime::now()
299            .duration_since(UNIX_EPOCH)
300            .map(|d| d.as_secs())
301            .unwrap_or(0)
302    }
303
304    fn sig_dir(&self, sig: &str) -> PathBuf {
305        self.root.join("stages").join(sig)
306    }
307    fn impl_dir(&self, sig: &str) -> PathBuf {
308        self.sig_dir(sig).join("implementations")
309    }
310    fn tests_dir(&self, sig: &str) -> PathBuf {
311        self.sig_dir(sig).join("tests")
312    }
313    fn specs_dir(&self, sig: &str) -> PathBuf {
314        self.sig_dir(sig).join("specs")
315    }
316    fn lifecycle_path(&self, sig: &str) -> PathBuf {
317        self.sig_dir(sig).join("lifecycle.json")
318    }
319
320    // ---- publish ----
321
322    /// Publish a stage as **Draft**. Returns the StageId.
323    /// Idempotent: republishing the same canonical AST returns the same
324    /// StageId without writing duplicates.
325    pub fn publish(&self, stage: &Stage) -> Result<String, StoreError> {
326        self.publish_signed(stage, None)
327    }
328
329    /// Like [`Self::publish`] but optionally attaches an Ed25519
330    /// signature over the StageId (#227). When `signer` is `Some`,
331    /// the persisted metadata gets a `signature` field that
332    /// downstream consumers can verify via
333    /// [`lex_vcs::verify_stage_id`].
334    ///
335    /// Idempotency: if a metadata file already exists the signature
336    /// is *not* re-written. This preserves "republishing is a no-op"
337    /// even across different signers — promoting a signed stage
338    /// requires a fresh stage hash anyway, so a metadata overwrite
339    /// would be the wrong primitive.
340    pub fn publish_signed(
341        &self,
342        stage: &Stage,
343        signer: Option<&lex_vcs::Keypair>,
344    ) -> Result<String, StoreError> {
345        let sig = sig_id(stage).ok_or(StoreError::CannotPublishImport)?;
346        let stage_id = stage_id(stage).ok_or(StoreError::CannotPublishImport)?;
347        let name = stage_name(stage).to_string();
348
349        fs::create_dir_all(self.impl_dir(&sig))?;
350        fs::create_dir_all(self.tests_dir(&sig))?;
351        fs::create_dir_all(self.specs_dir(&sig))?;
352
353        let ast_path = self.impl_dir(&sig).join(format!("{}.ast.json", stage_id));
354        let delta_path = self.impl_dir(&sig).join(format!("{}.delta.json", stage_id));
355        let meta_path = self
356            .impl_dir(&sig)
357            .join(format!("{}.metadata.json", stage_id));
358
359        // #261 slice 3: try delta encoding against the most recent
360        // prior stage in this sig's lifecycle. Falls back to a full
361        // snapshot when (a) no prior stage exists, (b) the diff
362        // ratio is over the threshold, or (c) the delta chain is
363        // already at its cap. The decision is internal — callers
364        // see the same `Stage` object on `get_ast` regardless.
365        if !ast_path.exists() && !delta_path.exists() {
366            self.persist_stage_bytes(&sig, &stage_id, stage, &ast_path, &delta_path)?;
367        }
368        if !meta_path.exists() {
369            let signature = signer.map(|kp| kp.sign_stage_id(&stage_id));
370            let metadata = Metadata {
371                stage_id: stage_id.clone(),
372                sig_id: sig.clone(),
373                name,
374                published_at: Self::now(),
375                note: None,
376                signature,
377            };
378            write_canonical_json(&meta_path, &metadata)?;
379        }
380
381        // Lifecycle: append a Draft transition for first publish.
382        let mut life = self.read_lifecycle(&sig).unwrap_or_else(|_| Lifecycle {
383            sig_id: sig.clone(),
384            ..Default::default()
385        });
386        if !life.transitions.iter().any(|t| t.stage_id == stage_id) {
387            life.transitions.push(Transition {
388                stage_id: stage_id.clone(),
389                from: StageStatus::Draft, // synthesized; "from" of first transition is itself
390                to: StageStatus::Draft,
391                at: Self::now(),
392                reason: None,
393            });
394            self.write_lifecycle(&sig, &life)?;
395        }
396        Ok(stage_id)
397    }
398
399    // ---- lifecycle ----
400
401    pub fn activate(&self, stage_id: &str) -> Result<(), StoreError> {
402        let (sig, mut life) = self.lookup_lifecycle(stage_id)?;
403        // Demote any currently-Active impls for this SigId to Deprecated.
404        let active = life.current_active().map(|s| s.to_string());
405        if let Some(prev) = active {
406            if prev != stage_id {
407                life.transitions.push(Transition {
408                    stage_id: prev,
409                    from: StageStatus::Active,
410                    to: StageStatus::Deprecated,
411                    at: Self::now(),
412                    reason: Some("superseded".into()),
413                });
414            }
415        }
416        let cur = life.status_of(stage_id);
417        if cur == Some(StageStatus::Tombstone) {
418            return Err(StoreError::InvalidTransition(
419                "tombstoned cannot be activated".into(),
420            ));
421        }
422        life.transitions.push(Transition {
423            stage_id: stage_id.into(),
424            from: cur.unwrap_or(StageStatus::Draft),
425            to: StageStatus::Active,
426            at: Self::now(),
427            reason: None,
428        });
429        self.write_lifecycle(&sig, &life)
430    }
431
432    pub fn deprecate(&self, stage_id: &str, reason: impl Into<String>) -> Result<(), StoreError> {
433        let (sig, mut life) = self.lookup_lifecycle(stage_id)?;
434        let cur = life
435            .status_of(stage_id)
436            .ok_or_else(|| StoreError::UnknownStage(stage_id.into()))?;
437        if cur != StageStatus::Active {
438            return Err(StoreError::InvalidTransition(format!(
439                "{cur:?} ⇒ Deprecated"
440            )));
441        }
442        life.transitions.push(Transition {
443            stage_id: stage_id.into(),
444            from: cur,
445            to: StageStatus::Deprecated,
446            at: Self::now(),
447            reason: Some(reason.into()),
448        });
449        self.write_lifecycle(&sig, &life)
450    }
451
452    pub fn tombstone(&self, stage_id: &str) -> Result<(), StoreError> {
453        let (sig, mut life) = self.lookup_lifecycle(stage_id)?;
454        let cur = life
455            .status_of(stage_id)
456            .ok_or_else(|| StoreError::UnknownStage(stage_id.into()))?;
457        if cur != StageStatus::Deprecated {
458            return Err(StoreError::InvalidTransition(format!(
459                "{cur:?} ⇒ Tombstone"
460            )));
461        }
462        life.transitions.push(Transition {
463            stage_id: stage_id.into(),
464            from: cur,
465            to: StageStatus::Tombstone,
466            at: Self::now(),
467            reason: None,
468        });
469        self.write_lifecycle(&sig, &life)
470    }
471
472    // ---- queries ----
473
474    /// The current Active StageId for a signature, or `None`.
475    pub fn resolve_sig(&self, sig: &str) -> Result<Option<String>, StoreError> {
476        let life = match self.read_lifecycle(sig) {
477            Ok(l) => l,
478            Err(_) => return Ok(None),
479        };
480        Ok(life.current_active().map(|s| s.to_string()))
481    }
482
483    /// Per-stage history for a SigId, ordered chronologically by
484    /// the *last* transition timestamp. Returns one entry per
485    /// distinct StageId that has ever been published under `sig`.
486    /// `Ok(vec![])` if the SigId doesn't exist in the store.
487    ///
488    /// Used by `lex blame` to render "where does this fn come from".
489    pub fn sig_history(&self, sig: &str) -> Result<Vec<StageHistoryEntry>, StoreError> {
490        let life = match self.read_lifecycle(sig) {
491            Ok(l) => l,
492            Err(_) => return Ok(Vec::new()),
493        };
494        // Collapse transitions: latest status + last_at per stage,
495        // plus the timestamp of the first Draft transition (≈ when
496        // the stage was published) when one exists.
497        let mut by_stage: indexmap::IndexMap<String, StageHistoryEntry> = indexmap::IndexMap::new();
498        for t in &life.transitions {
499            let entry = by_stage
500                .entry(t.stage_id.clone())
501                .or_insert(StageHistoryEntry {
502                    stage_id: t.stage_id.clone(),
503                    status: t.to,
504                    last_at: t.at,
505                    published_at: None,
506                });
507            entry.status = t.to;
508            entry.last_at = t.at;
509            if t.from == StageStatus::Draft && entry.published_at.is_none() {
510                entry.published_at = Some(t.at);
511            }
512            if t.to == StageStatus::Draft && entry.published_at.is_none() {
513                // Initial publication: Draft is the *destination*.
514                entry.published_at = Some(t.at);
515            }
516        }
517        let mut out: Vec<StageHistoryEntry> = by_stage.into_values().collect();
518        // Sort newest first so `lex blame` shows recent activity at top.
519        out.sort_by_key(|e| std::cmp::Reverse(e.last_at));
520        Ok(out)
521    }
522
523    pub fn get_ast(&self, stage_id: &str) -> Result<Stage, StoreError> {
524        let (sig, _) = self.lookup_lifecycle(stage_id)?;
525        let bytes = self.read_stage_canonical_bytes(&sig, stage_id)?;
526        Ok(serde_json::from_slice(&bytes)?)
527    }
528
529    /// Read the canonical bytes of a stage, walking back through
530    /// any delta chain (#261 slice 3). The recursion ends at a
531    /// `<stage_id>.ast.json` file (a full snapshot) or, in the
532    /// degenerate case of a missing chain, with `UnknownStage`.
533    fn read_stage_canonical_bytes(&self, sig: &str, stage_id: &str) -> Result<Vec<u8>, StoreError> {
534        let ast_path = self.impl_dir(sig).join(format!("{}.ast.json", stage_id));
535        if ast_path.exists() {
536            return Ok(fs::read(&ast_path)?);
537        }
538        let delta_path = self.impl_dir(sig).join(format!("{}.delta.json", stage_id));
539        if !delta_path.exists() {
540            return Err(StoreError::UnknownStage(stage_id.into()));
541        }
542        let delta_bytes = fs::read(&delta_path)?;
543        let delta: crate::delta::StageDelta = serde_json::from_slice(&delta_bytes)?;
544        let base_bytes = self.read_stage_canonical_bytes(sig, &delta.base_stage_id)?;
545        crate::delta::apply(&base_bytes, &delta).map_err(|e| {
546            StoreError::Io(std::io::Error::new(
547                std::io::ErrorKind::InvalidData,
548                format!("applying delta for {stage_id}: {e}"),
549            ))
550        })
551    }
552
553    /// Persist a freshly-published stage's canonical bytes (#261
554    /// slice 3). Tries delta encoding against the most recent
555    /// prior stage in the sig's lifecycle; falls back to a full
556    /// snapshot when no base exists, the diff ratio is too high,
557    /// or the delta chain is already at its cap.
558    fn persist_stage_bytes(
559        &self,
560        sig: &str,
561        stage_id: &str,
562        stage: &Stage,
563        ast_path: &Path,
564        delta_path: &Path,
565    ) -> Result<(), StoreError> {
566        let new_bytes = canonical_bytes(stage)?;
567        if let Some((base_stage_id, base_chain_length)) = self.pick_delta_base(sig, stage_id)? {
568            let base_bytes = self.read_stage_canonical_bytes(sig, &base_stage_id)?;
569            let (prefix, suffix, middle) = crate::delta::splice(&base_bytes, &new_bytes);
570            let chain_length = base_chain_length + 1;
571            if crate::delta::is_worth_encoding(middle.len(), new_bytes.len(), chain_length) {
572                let delta = crate::delta::StageDelta {
573                    base_stage_id,
574                    chain_length,
575                    common_prefix: prefix,
576                    common_suffix: suffix,
577                    middle_hex: hex::encode(&middle),
578                };
579                write_canonical_json(delta_path, &delta)?;
580                return Ok(());
581            }
582        }
583        // Fall through: full snapshot.
584        if let Some(parent) = ast_path.parent() {
585            fs::create_dir_all(parent)?;
586        }
587        fs::write(ast_path, &new_bytes)?;
588        Ok(())
589    }
590
591    /// Pick a base stage for delta encoding from the given sig's
592    /// lifecycle. Returns `(base_stage_id, base_chain_length)` for
593    /// the most-recent non-tombstoned prior stage, or `None` when
594    /// there is no candidate. The chain length is read off the
595    /// base's `.delta.json` (if any) to enforce the cap.
596    fn pick_delta_base(
597        &self,
598        sig: &str,
599        new_stage_id: &str,
600    ) -> Result<Option<(String, usize)>, StoreError> {
601        let life = self.read_lifecycle(sig).ok();
602        let Some(life) = life else {
603            return Ok(None);
604        };
605        // Walk transitions newest-first; pick the first prior
606        // stage that isn't this one and isn't tombstoned.
607        let mut latest_per_stage: indexmap::IndexMap<&str, StageStatus> = indexmap::IndexMap::new();
608        for t in &life.transitions {
609            latest_per_stage.insert(&t.stage_id, t.to);
610        }
611        let mut candidates: Vec<&str> = latest_per_stage
612            .iter()
613            .filter(|(id, status)| **id != new_stage_id && **status != StageStatus::Tombstone)
614            .map(|(id, _)| *id)
615            .collect();
616        // Reverse to get newest-first (transitions are append-only,
617        // so latest_per_stage's iteration order matches insertion
618        // order, oldest-first).
619        candidates.reverse();
620        let Some(&base) = candidates.first() else {
621            return Ok(None);
622        };
623        let base_chain_length = self.delta_chain_length(sig, base)?;
624        Ok(Some((base.to_string(), base_chain_length)))
625    }
626
627    /// Length of the delta chain ending at `stage_id`. Zero when
628    /// the stage is a full snapshot (`.ast.json` present); the
629    /// stored `chain_length` from `.delta.json` otherwise.
630    fn delta_chain_length(&self, sig: &str, stage_id: &str) -> Result<usize, StoreError> {
631        let ast_path = self.impl_dir(sig).join(format!("{}.ast.json", stage_id));
632        if ast_path.exists() {
633            return Ok(0);
634        }
635        let delta_path = self.impl_dir(sig).join(format!("{}.delta.json", stage_id));
636        if !delta_path.exists() {
637            return Ok(0);
638        }
639        let bytes = fs::read(&delta_path)?;
640        let delta: crate::delta::StageDelta = serde_json::from_slice(&bytes)?;
641        Ok(delta.chain_length)
642    }
643
644    pub fn get_metadata(&self, stage_id: &str) -> Result<Metadata, StoreError> {
645        let (sig, _) = self.lookup_lifecycle(stage_id)?;
646        let path = self
647            .impl_dir(&sig)
648            .join(format!("{}.metadata.json", stage_id));
649        let bytes = fs::read(&path)?;
650        Ok(serde_json::from_slice(&bytes)?)
651    }
652
653    pub fn get_status(&self, stage_id: &str) -> Result<StageStatus, StoreError> {
654        let (_sig, life) = self.lookup_lifecycle(stage_id)?;
655        life.status_of(stage_id)
656            .ok_or_else(|| StoreError::UnknownStage(stage_id.into()))
657    }
658
659    pub fn list_stages_by_name(&self, name: &str) -> Result<Vec<String>, StoreError> {
660        // Walk every SigId → check metadata of any implementation; if its
661        // name matches, include the SigId.
662        let mut out = Vec::new();
663        let stages_dir = self.root.join("stages");
664        if !stages_dir.exists() {
665            return Ok(out);
666        }
667        for entry in fs::read_dir(&stages_dir)? {
668            let entry = entry?;
669            let sig_dir = entry.path();
670            if !sig_dir.is_dir() {
671                continue;
672            }
673            let sig = entry.file_name().to_string_lossy().to_string();
674            // Look at any one metadata file under this SigId.
675            let impls = self.impl_dir(&sig);
676            if !impls.exists() {
677                continue;
678            }
679            for f in fs::read_dir(impls)? {
680                let f = f?;
681                let p = f.path();
682                if p.extension().is_some_and(|e| e == "json")
683                    && p.file_name()
684                        .is_some_and(|n| n.to_string_lossy().ends_with(".metadata.json"))
685                {
686                    if let Ok(bytes) = fs::read(&p) {
687                        if let Ok(m) = serde_json::from_slice::<Metadata>(&bytes) {
688                            if m.name == name {
689                                if !out.contains(&sig) {
690                                    out.push(sig.clone());
691                                }
692                                break;
693                            }
694                        }
695                    }
696                }
697            }
698        }
699        out.sort();
700        Ok(out)
701    }
702
703    pub fn list_sigs(&self) -> Result<Vec<String>, StoreError> {
704        let stages_dir = self.root.join("stages");
705        let mut out = Vec::new();
706        if !stages_dir.exists() {
707            return Ok(out);
708        }
709        for entry in fs::read_dir(stages_dir)? {
710            let entry = entry?;
711            if entry.file_type()?.is_dir() {
712                out.push(entry.file_name().to_string_lossy().to_string());
713            }
714        }
715        out.sort();
716        Ok(out)
717    }
718
719    // ---- tests/specs as metadata (§4.4) ----
720
721    pub fn attach_test(&self, sig: &str, test: &Test) -> Result<String, StoreError> {
722        if !self.sig_dir(sig).exists() {
723            return Err(StoreError::UnknownSig(sig.into()));
724        }
725        fs::create_dir_all(self.tests_dir(sig))?;
726        let path = self.tests_dir(sig).join(format!("{}.json", test.id));
727        write_canonical_json(&path, test)?;
728        Ok(test.id.clone())
729    }
730
731    pub fn list_tests(&self, sig: &str) -> Result<Vec<Test>, StoreError> {
732        let dir = self.tests_dir(sig);
733        if !dir.exists() {
734            return Ok(Vec::new());
735        }
736        let mut out = Vec::new();
737        for f in fs::read_dir(dir)? {
738            let f = f?;
739            if f.path().extension().is_some_and(|e| e == "json") {
740                let bytes = fs::read(f.path())?;
741                out.push(serde_json::from_slice(&bytes)?);
742            }
743        }
744        Ok(out)
745    }
746
747    pub fn attach_spec(&self, sig: &str, spec: &Spec) -> Result<String, StoreError> {
748        if !self.sig_dir(sig).exists() {
749            return Err(StoreError::UnknownSig(sig.into()));
750        }
751        fs::create_dir_all(self.specs_dir(sig))?;
752        let path = self.specs_dir(sig).join(format!("{}.json", spec.id));
753        write_canonical_json(&path, spec)?;
754        Ok(spec.id.clone())
755    }
756
757    pub fn list_specs(&self, sig: &str) -> Result<Vec<Spec>, StoreError> {
758        let dir = self.specs_dir(sig);
759        if !dir.exists() {
760            return Ok(Vec::new());
761        }
762        let mut out = Vec::new();
763        for f in fs::read_dir(dir)? {
764            let f = f?;
765            if f.path().extension().is_some_and(|e| e == "json") {
766                let bytes = fs::read(f.path())?;
767                out.push(serde_json::from_slice(&bytes)?);
768            }
769        }
770        Ok(out)
771    }
772
773    // ---- traces (§4.2 / M7) ----
774
775    // Native run-trace store — gated behind the `trace` feature (depends on
776    // lex-trace). Off when a lower crate (lex-runtime) depends on lex-store to
777    // avoid a dependency cycle; the blob/stage store below is unaffected.
778    #[cfg(feature = "trace")]
779    fn trace_path(&self, run_id: &str) -> PathBuf {
780        self.root.join("traces").join(run_id).join("trace.json")
781    }
782
783    #[cfg(feature = "trace")]
784    pub fn save_trace(&self, tree: &lex_trace::TraceTree) -> Result<String, StoreError> {
785        let path = self.trace_path(&tree.run_id);
786        write_canonical_json(&path, tree)?;
787        Ok(tree.run_id.clone())
788    }
789
790    #[cfg(feature = "trace")]
791    pub fn load_trace(&self, run_id: &str) -> Result<lex_trace::TraceTree, StoreError> {
792        let bytes = fs::read(self.trace_path(run_id))?;
793        Ok(serde_json::from_slice(&bytes)?)
794    }
795
796    pub fn list_traces(&self) -> Result<Vec<String>, StoreError> {
797        let dir = self.root.join("traces");
798        if !dir.exists() {
799            return Ok(Vec::new());
800        }
801        let mut out = Vec::new();
802        for entry in fs::read_dir(dir)? {
803            let entry = entry?;
804            if entry.file_type()?.is_dir() {
805                out.push(entry.file_name().to_string_lossy().to_string());
806            }
807        }
808        out.sort();
809        Ok(out)
810    }
811
812    // ---- internals ----
813
814    fn lookup_lifecycle(&self, stage_id: &str) -> Result<(String, Lifecycle), StoreError> {
815        // Walk every SigId, find which one contains this StageId.
816        for sig in self.list_sigs()? {
817            if let Ok(life) = self.read_lifecycle(&sig) {
818                if life.transitions.iter().any(|t| t.stage_id == stage_id) {
819                    return Ok((sig, life));
820                }
821            }
822        }
823        Err(StoreError::UnknownStage(stage_id.into()))
824    }
825
826    fn read_lifecycle(&self, sig: &str) -> Result<Lifecycle, StoreError> {
827        let path = self.lifecycle_path(sig);
828        if !path.exists() {
829            return Ok(Lifecycle {
830                sig_id: sig.into(),
831                transitions: Vec::new(),
832            });
833        }
834        let bytes = fs::read(&path)?;
835        Ok(serde_json::from_slice(&bytes)?)
836    }
837
838    fn write_lifecycle(&self, sig: &str, life: &Lifecycle) -> Result<(), StoreError> {
839        write_canonical_json(&self.lifecycle_path(sig), life)
840    }
841
842    /// Apply a published program to a branch as a sequence of typed
843    /// operations. Returns the ordered list of op_ids + the new
844    /// head_op. The caller (`lex publish` CLI, `lex serve`'s HTTP
845    /// handler) is responsible for computing the `DiffReport` against
846    /// the current branch head — the diff infrastructure lives in
847    /// `lex-vcs::compute_diff` (previously `lex-cli`) to keep this
848    /// layer from owning diffing logic.
849    ///
850    /// On success: every op in the returned list is durable in the
851    /// op log and the branch's head_op points at the last one.
852    /// On a no-op (no diff): returns empty `ops` and the existing
853    /// `head_op` unchanged.
854    pub fn publish_program(
855        &self,
856        branch: &str,
857        stages: &[lex_ast::Stage],
858        diff: &lex_vcs::DiffReport,
859        new_imports: &lex_vcs::ImportMap,
860        activate: bool,
861    ) -> Result<PublishOutcome, StoreError> {
862        self.publish_program_signed(branch, stages, diff, new_imports, activate, None)
863    }
864
865    /// Signed variant of [`Self::publish_program`] (#227). Every
866    /// stage written under this batch gets the same signer; per-stage
867    /// keys aren't supported because the agent identity model treats
868    /// a publish as a single authorial act.
869    pub fn publish_program_signed(
870        &self,
871        branch: &str,
872        stages: &[lex_ast::Stage],
873        diff: &lex_vcs::DiffReport,
874        new_imports: &lex_vcs::ImportMap,
875        activate: bool,
876        signer: Option<&lex_vcs::Keypair>,
877    ) -> Result<PublishOutcome, StoreError> {
878        use std::collections::{BTreeMap, BTreeSet};
879
880        // #130's write-time gate: verify the candidate program
881        // typechecks (and effects are correctly declared) before
882        // any disk side-effect. If anything fails, return the
883        // structured envelope and leave the branch head unchanged
884        // — the store's "always-valid HEAD" invariant only holds
885        // because this is the only batch-publish path that
886        // advances heads. Single-op writes via the lower-level
887        // `apply_operation` are not gated yet (#130 follow-up).
888        if let Err(errors) = lex_types::check_program(stages) {
889            return Err(StoreError::TypeError(errors));
890        }
891
892        // Build old-side views from the current branch.
893        let old_head = self.branch_head(branch)?;
894        let old_name_to_sig: BTreeMap<String, String> = old_head
895            .iter()
896            .filter_map(|(sig, stg)| self.get_metadata(stg).ok().map(|m| (m.name, sig.clone())))
897            .collect();
898        let old_effects: BTreeMap<String, BTreeSet<String>> = old_head
899            .iter()
900            .filter_map(|(sig, stg)| {
901                let ast = self.get_ast(stg).ok()?;
902                match ast {
903                    lex_ast::Stage::FnDecl(fd) => {
904                        let s: BTreeSet<String> =
905                            fd.effects.iter().map(|e| e.name.clone()).collect();
906                        Some((sig.clone(), s))
907                    }
908                    _ => None,
909                }
910            })
911            .collect();
912        let old_imports = self.derive_imports_from_oplog(branch)?;
913
914        let op_kinds = lex_vcs::diff_to_ops(lex_vcs::DiffInputs {
915            old_head: &old_head,
916            old_name_to_sig: &old_name_to_sig,
917            old_effects: &old_effects,
918            old_imports: &old_imports,
919            new_stages: stages,
920            new_imports,
921            diff,
922        })
923        .map_err(|e| StoreError::InvalidTransition(format!("diff_to_ops: {e}")))?;
924
925        let mut ops_out: Vec<PublishOp> = Vec::new();
926        let mut last_op_id: Option<lex_vcs::OpId> = None;
927        for kind in op_kinds {
928            // Persist the underlying stage AST/metadata if this op
929            // produces or replaces one.
930            if let Some(stg) = stage_for_kind(&kind, stages) {
931                if !matches!(stg, lex_ast::Stage::Import(_)) {
932                    self.publish_signed(stg, signer)?;
933                    if activate {
934                        if let Some(stage_id_str) = stage_id(stg) {
935                            let _ = self.activate(&stage_id_str);
936                        }
937                    }
938                }
939            }
940            let transition = transition_for_kind(&kind);
941            let attestable = attestable_stage_ids(&transition);
942            let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
943            let op =
944                lex_vcs::Operation::new(kind.clone(), head_now.into_iter().collect::<Vec<_>>());
945            let op_id = self.apply_operation(branch, op, transition)?;
946            self.record_typecheck_passed(&attestable, &op_id)?;
947            ops_out.push(PublishOp {
948                op_id: op_id.clone(),
949                kind: serde_json::to_value(&kind).map_err(StoreError::Serde)?,
950            });
951            last_op_id = Some(op_id);
952        }
953
954        let head_op = match last_op_id {
955            Some(id) => Some(id),
956            // No ops applied; return whatever the head was already.
957            None => self.get_branch(branch)?.and_then(|b| b.head_op),
958        };
959
960        Ok(PublishOutcome {
961            ops: ops_out,
962            head_op,
963        })
964    }
965
966    pub fn derive_imports_from_oplog(
967        &self,
968        branch: &str,
969    ) -> Result<lex_vcs::ImportMap, StoreError> {
970        use lex_vcs::OperationKind::*;
971        let log = lex_vcs::OpLog::open(self.root())?;
972        let head = match self.get_branch(branch)?.and_then(|b| b.head_op) {
973            Some(h) => h,
974            None => return Ok(Default::default()),
975        };
976        let mut out: lex_vcs::ImportMap = Default::default();
977        for r in log.walk_forward(&head, None)? {
978            match r.op.kind {
979                AddImport { in_file, module } => {
980                    out.entry(in_file).or_default().insert(module);
981                }
982                RemoveImport { in_file, module } => {
983                    if let Some(set) = out.get_mut(&in_file) {
984                        set.remove(&module);
985                    }
986                }
987                _ => {}
988            }
989        }
990        Ok(out)
991    }
992
993    /// Apply an operation to a branch and advance its head_op.
994    ///
995    /// The single advance path. Validates parents via `lex_vcs::apply`,
996    /// persists the operation via the op log, then atomically advances
997    /// the branch file's head_op via `set_branch_head_op`.
998    ///
999    /// Errors:
1000    /// - `UnknownBranch`: branch does not exist (no op is persisted).
1001    /// - `Apply(ApplyError::StaleParent)`: the op's parents don't
1002    ///   match the branch head — head is unchanged. Callers that
1003    ///   want retry-on-stale (e.g. `lex publish` re-running against
1004    ///   a moved head) match on this variant explicitly.
1005    /// - `Apply(ApplyError::UnknownMergeParent)`: a merge op's
1006    ///   second parent isn't in the log.
1007    /// - `Io`: filesystem error during persist or branch advance.
1008    ///
1009    /// Crash recovery: between op persist and branch advance, a crash
1010    /// can leave an orphan op record in the log with no branch
1011    /// pointing at it. The op is content-addressed and cheap to
1012    /// re-derive from the same source. See
1013    /// Apply a single op against `branch`, gated on the candidate
1014    /// program typechecking. The per-op variant of #130's
1015    /// write-time gate — counterpart to [`Self::publish_program`]'s
1016    /// batch-mode check.
1017    ///
1018    /// `candidate` is the sequence of `Stage`s that *would* exist
1019    /// on this branch after the op is applied. Caller's
1020    /// responsibility: today neither `lex-store` nor `lex-vcs`
1021    /// reconstruct the candidate from the op + branch state on
1022    /// behalf of the caller. The natural callers (HTTP `POST
1023    /// /v1/publish` for a single op; agent harnesses driving
1024    /// merges via the future #134 API) already have the candidate
1025    /// in memory.
1026    ///
1027    /// On rejection: branch head unchanged, no op record persisted.
1028    /// Same atomicity guarantee as the publish path.
1029    ///
1030    /// # Why a separate method, not a flag on `apply_operation`
1031    ///
1032    /// The merge engine in `lex-vcs::merge` calls
1033    /// `Store::apply_operation` directly to land merge ops, and at
1034    /// merge time the resolved program isn't a single `Vec<Stage>`
1035    /// the way it is on the publish path — it's a per-sig
1036    /// resolution map. Forcing a candidate through `apply_operation`
1037    /// would either require the merge engine to assemble one (slow,
1038    /// every active stage off disk) or accept `Option<&[Stage]>`
1039    /// and silently skip the gate — the second is exactly the kind
1040    /// of "secretly opt-out" path #130 is trying to remove. The
1041    /// honest split is two methods: `apply_operation` for callers
1042    /// that already typecheck their inputs (or don't need to —
1043    /// rare, but the merge-resolve case), `apply_operation_checked`
1044    /// for everyone else.
1045    pub fn apply_operation_checked(
1046        &self,
1047        branch: &str,
1048        op: lex_vcs::Operation,
1049        transition: lex_vcs::StageTransition,
1050        candidate: &[lex_ast::Stage],
1051    ) -> Result<lex_vcs::OpId, StoreError> {
1052        if let Err(errors) = lex_types::check_program(candidate) {
1053            // #281: emit a `RepairHint` attestation against each
1054            // candidate stage the transition was about to produce.
1055            // The op record itself isn't persisted (the gate is
1056            // pre-persistence), but the candidate stage IS — the
1057            // transform-flow methods publish before this call.
1058            // The attached hint lets `lex repair <op_id>` and
1059            // future LLM-assisted apply paths read the structured
1060            // errors without re-running the typecheck.
1061            let attestable = attestable_stage_ids(&transition);
1062            let failed_op_id = op.op_id();
1063            let _ = self.record_repair_hint(&attestable, &failed_op_id, &errors);
1064            return Err(StoreError::TypeError(errors));
1065        }
1066        // #292 slice 3: per-session budget gate. After typecheck
1067        // passes, refuse the op if it would push its session's
1068        // monotonic spend over the configured cap. Sessions
1069        // without an intent_id, or with an intent whose session
1070        // has no cap configured, sail through.
1071        self.check_session_budget(&op)?;
1072        let attestable = attestable_stage_ids(&transition);
1073        let op_effects = op_declared_effects(&op.kind);
1074        // #262: CAS retry loop. Single-parent ops can be safely
1075        // re-persisted under a new parent on contention (the kind
1076        // is invariant; only `parents` changes). Merge ops (already
1077        // 2-parent) come through the merge engine which has its own
1078        // coordination; we don't retry them here — we'll see the
1079        // first attempt's CAS fail and surface Contention.
1080        self.cas_retry_advance(branch, op, transition, |new_head| {
1081            self.record_typecheck_passed(&attestable, &new_head.op_id)?;
1082            self.run_required_attestations_gate(branch, &new_head.op_id, &attestable, &op_effects)
1083        })
1084    }
1085
1086    /// Open the attestation log rooted at this store. The log lives
1087    /// under `<root>/attestations/`; opening is idempotent and cheap
1088    /// (`fs::create_dir_all`). Exposed publicly so consumers — `lex
1089    /// blame --with-evidence`, `GET /v1/stage/<id>/attestations` —
1090    /// can read what the store gate emitted without round-tripping
1091    /// through this crate's API surface.
1092    /// Recompute a producer's trust score from its recent
1093    /// attestation history and emit a fresh `ProducerTrust`
1094    /// attestation (#293). Score = `passed / (passed + failed
1095    /// + inconclusive)` over the last `window` attestations
1096    /// produced by `tool_id`, expressed in thousandths
1097    /// (`0..=1000`).
1098    ///
1099    /// Refuses to grant trust when the tool has an active
1100    /// `ProducerBlock` — the block wins as a hard veto. Returns
1101    /// `Ok(None)` for "no attestations to score" (a brand-new
1102    /// producer); the caller can choose how to handle it
1103    /// (typically: skip the publish until evidence accrues).
1104    ///
1105    /// `granted_by` is the identity of the actor running the
1106    /// recompute (typically the human admin, or "lex-ci-bot"
1107    /// for an automated nightly).
1108    pub fn recompute_producer_trust(
1109        &self,
1110        tool_id: &str,
1111        window: usize,
1112        granted_by: &str,
1113    ) -> Result<Option<lex_vcs::AttestationId>, StoreError> {
1114        let log = self.attestation_log()?;
1115        let all = log.list_all()?;
1116        // Hard veto: don't grant trust to a blocked tool.
1117        if lex_vcs::active_producer_block(&all, tool_id).is_some() {
1118            return Err(StoreError::InvalidTransition(format!(
1119                "cannot recompute trust for `{tool_id}` — \
1120                 producer is currently blocked"
1121            )));
1122        }
1123        // Filter to attestations from this tool, newest-first by
1124        // timestamp, then take the window.
1125        let mut from_tool: Vec<&lex_vcs::Attestation> = all
1126            .iter()
1127            .filter(|a| a.produced_by.tool == tool_id)
1128            // Ignore self-referential trust attestations (we're
1129            // scoring evidence, not previous trust statements).
1130            .filter(|a| {
1131                !matches!(
1132                    a.kind,
1133                    lex_vcs::AttestationKind::ProducerTrust { .. }
1134                        | lex_vcs::AttestationKind::TrustWaived { .. }
1135                )
1136            })
1137            .collect();
1138        from_tool.sort_by_key(|a| std::cmp::Reverse(a.timestamp));
1139        from_tool.truncate(window);
1140        if from_tool.is_empty() {
1141            return Ok(None);
1142        }
1143        let (mut passed, mut total) = (0u64, 0u64);
1144        for a in &from_tool {
1145            total += 1;
1146            if matches!(a.result, lex_vcs::AttestationResult::Passed) {
1147                passed += 1;
1148            }
1149        }
1150        let score = if total == 0 {
1151            0
1152        } else {
1153            let raw = (passed as f64) * 1000.0 / (total as f64);
1154            raw.round().clamp(0.0, 1000.0) as u32
1155        };
1156        let head_op = self
1157            .list_branches()?
1158            .into_iter()
1159            .find_map(|b| self.get_branch(&b).ok().flatten().and_then(|x| x.head_op))
1160            .unwrap_or_else(|| "fresh".into());
1161        let evidence = format!(
1162            "window={window}, sample={}, head_op={head_op:.16}",
1163            from_tool.len()
1164        );
1165        let attestation = lex_vcs::Attestation::new(
1166            tool_id.to_string(),
1167            None,
1168            None,
1169            lex_vcs::AttestationKind::ProducerTrust {
1170                tool_id: tool_id.into(),
1171                score_thousandths: score,
1172                evidence,
1173                granted_by: granted_by.into(),
1174            },
1175            lex_vcs::AttestationResult::Passed,
1176            producer_trust_producer(),
1177            None,
1178        );
1179        let id = attestation.attestation_id.clone();
1180        log.put(&attestation)?;
1181        Ok(Some(id))
1182    }
1183
1184    /// The latest live `ProducerTrust` score (thousandths, `0..=1000`) for
1185    /// every producer that currently has trust: the newest score per tool by
1186    /// timestamp, excluding any tool under an active `ProducerBlock` (a block
1187    /// is a hard veto over trust, matching `recompute_producer_trust`).
1188    ///
1189    /// Used to export a capsule trusted-keys keyring from *earned* trust — the
1190    /// producer id doubles as the publisher's signing key downstream, so this
1191    /// turns track record into the allowlist `capsule install` consumes.
1192    pub fn live_producer_trust_scores(
1193        &self,
1194    ) -> Result<std::collections::BTreeMap<String, u32>, StoreError> {
1195        let log = self.attestation_log()?;
1196        let all = log.list_all()?;
1197        // Newest score per tool.
1198        let mut latest: std::collections::BTreeMap<String, (u64, u32)> =
1199            std::collections::BTreeMap::new();
1200        for a in &all {
1201            if let lex_vcs::AttestationKind::ProducerTrust {
1202                tool_id,
1203                score_thousandths,
1204                ..
1205            } = &a.kind
1206            {
1207                let entry = latest.entry(tool_id.clone()).or_insert((0, 0));
1208                if a.timestamp >= entry.0 {
1209                    *entry = (a.timestamp, *score_thousandths);
1210                }
1211            }
1212        }
1213        // Drop blocked producers; a block vetoes trust.
1214        let mut scores = std::collections::BTreeMap::new();
1215        for (tool, (_, score)) in latest {
1216            if lex_vcs::active_producer_block(&all, &tool).is_some() {
1217                continue;
1218            }
1219            scores.insert(tool, score);
1220        }
1221        Ok(scores)
1222    }
1223
1224    pub fn attestation_log(&self) -> Result<lex_vcs::AttestationLog, StoreError> {
1225        Ok(lex_vcs::AttestationLog::open(self.root())?)
1226    }
1227
1228    /// Emit one `TypeCheck::Passed` attestation per stage produced by
1229    /// a successful gated apply. Idempotent on `attestation_id` —
1230    /// re-running the same gate run dedups via content addressing.
1231    ///
1232    /// Failure modes: `io::Error` from the attestation log (disk
1233    /// full, perms). The op has already landed by the time this
1234    /// runs; an error here means the op is durable but the evidence
1235    /// is missing. We propagate so the caller sees the partial
1236    /// state rather than silently swallowing — re-attesting the
1237    /// same op against the same op_id is idempotent (content
1238    /// addressing) so a retry is safe once the underlying issue is
1239    /// fixed.
1240    fn record_typecheck_passed(
1241        &self,
1242        stage_ids: &[String],
1243        op_id: &lex_vcs::OpId,
1244    ) -> Result<(), StoreError> {
1245        if stage_ids.is_empty() {
1246            return Ok(());
1247        }
1248        let log = self.attestation_log()?;
1249        for stage_id in stage_ids {
1250            let attestation = lex_vcs::Attestation::new(
1251                stage_id.clone(),
1252                Some(op_id.clone()),
1253                None,
1254                lex_vcs::AttestationKind::TypeCheck,
1255                lex_vcs::AttestationResult::Passed,
1256                typecheck_producer(),
1257                None,
1258            );
1259            log.put(&attestation)?;
1260        }
1261        Ok(())
1262    }
1263
1264    /// Consult `policy.session_budgets` for the op's session
1265    /// (resolved via `op.intent_id → Intent.session_id`) and
1266    /// refuse if applying would push the session's monotonic spend
1267    /// over the configured cap (#292 slice 3).
1268    ///
1269    /// Ops without an `intent_id`, or whose intent has no
1270    /// configured cap, return Ok without any disk read.
1271    fn check_session_budget(&self, op: &lex_vcs::Operation) -> Result<(), StoreError> {
1272        let Some(intent_id) = op.intent_id.as_deref() else {
1273            return Ok(());
1274        };
1275        let intent_log = lex_vcs::IntentLog::open(self.root())?;
1276        let Some(intent) = intent_log.get(&intent_id.to_string())? else {
1277            // Dangling intent — treat as "no session" and let it
1278            // sail through. Slice 1's ledger already documents
1279            // this as graceful-degradation semantics.
1280            return Ok(());
1281        };
1282        let policy = crate::policy::load(self.root())?.unwrap_or_default();
1283        let Some(cap) = policy.session_budgets.cap_for(&intent.session_id) else {
1284            return Ok(());
1285        };
1286        // Recompute the session's current spend + the contribution
1287        // from this op. Re-running the ledger walk on every gated
1288        // op is O(branch history); see #292 slice 1's note about
1289        // a future on-disk cache.
1290        let current = self.session_budget(&intent.session_id)?;
1291        let increment = crate::budget::monotonic_spend_of(&op.kind);
1292        let spent_after = current.spent.saturating_add(increment);
1293        if spent_after > cap {
1294            return Err(StoreError::BudgetExceeded {
1295                session_id: intent.session_id,
1296                cap,
1297                spent_after,
1298            });
1299        }
1300        Ok(())
1301    }
1302
1303    /// Emit `RepairHint` attestations for a TypeError-rejected op
1304    /// (#281). One per candidate stage in the transition. The hint
1305    /// records the *would-be* op_id (deterministic, content-
1306    /// addressed even though the op record was never persisted)
1307    /// and the structured errors.
1308    ///
1309    /// #306 slice 3: `suggested_transform` is populated from the
1310    /// static (rule_tag → likely_transform) table for the *first*
1311    /// error in the batch. The LLM-driven `lex repair --apply`
1312    /// flow can still overwrite this with a higher-quality
1313    /// suggestion; the static value is the floor, not the ceiling.
1314    ///
1315    /// Best-effort: a write failure here is swallowed by the
1316    /// caller (the original `TypeError` is the load-bearing
1317    /// signal; missing the hint is recoverable on a retry).
1318    fn record_repair_hint(
1319        &self,
1320        stage_ids: &[String],
1321        failed_op_id: &lex_vcs::OpId,
1322        errors: &[lex_types::TypeError],
1323    ) -> Result<(), StoreError> {
1324        if stage_ids.is_empty() {
1325            return Ok(());
1326        }
1327        let errors_json = serde_json::to_value(errors).map_err(StoreError::Serde)?;
1328        // #306 slice 3: look up the static suggested_transform for
1329        // the first error's rule_tag. Multiple errors per op are
1330        // possible — when they fire in lockstep (e.g. one bad let
1331        // binding propagates to several use sites), the first
1332        // error's rule_tag is usually the load-bearing one to fix.
1333        let suggested_transform = errors
1334            .first()
1335            .and_then(|e| lex_types::suggested_transform_for(e.rule_tag()));
1336        let log = self.attestation_log()?;
1337        for stage_id in stage_ids {
1338            let attestation = lex_vcs::Attestation::new(
1339                stage_id.clone(),
1340                None, // the failed op was never persisted; not the
1341                // attestation's op_id (which is for a
1342                // *successful* op).
1343                None,
1344                lex_vcs::AttestationKind::RepairHint {
1345                    failed_op_id: failed_op_id.clone(),
1346                    errors: errors_json.clone(),
1347                    suggested_transform: suggested_transform.clone(),
1348                },
1349                lex_vcs::AttestationResult::Failed {
1350                    detail: format!(
1351                        "op {} rejected: {} type error(s)",
1352                        failed_op_id,
1353                        errors.len()
1354                    ),
1355                },
1356                repair_hint_producer(),
1357                None,
1358            );
1359            log.put(&attestation)?;
1360        }
1361        Ok(())
1362    }
1363
1364    /// Emit `Trace` attestations linking an already-committed `op`
1365    /// to the run that produced it (#257). One attestation per
1366    /// produced stage (matching the `TypeCheck` emission contract
1367    /// — see [`Self::apply_operation_checked`]) with
1368    /// `op_id: Some(op_id)` set, so `lex trace --op <op_id>`
1369    /// surfaces the run.
1370    ///
1371    /// Returns the number of attestations emitted (zero for ops
1372    /// that produce no attestable stage, e.g. `Remove` /
1373    /// `ImportOnly`).
1374    ///
1375    /// Idempotent: re-emitting for the same
1376    /// `(run_id, root_target, op_id, stage_id, producer, result)`
1377    /// tuple dedups via content addressing.
1378    ///
1379    /// `op_id` must already exist in the op log — an unknown op
1380    /// surfaces as `StoreError::UnknownOp`.
1381    pub fn record_op_trace(
1382        &self,
1383        run_id: &str,
1384        root_target: &str,
1385        op_id: &lex_vcs::OpId,
1386        result: lex_vcs::AttestationResult,
1387        producer: lex_vcs::ProducerDescriptor,
1388    ) -> Result<usize, StoreError> {
1389        let log = lex_vcs::OpLog::open(self.root())?;
1390        let rec = log
1391            .get(op_id)?
1392            .ok_or_else(|| StoreError::UnknownOp(op_id.clone()))?;
1393        let stage_ids = attestable_stage_ids(&rec.produces);
1394        if stage_ids.is_empty() {
1395            return Ok(0);
1396        }
1397        let attlog = self.attestation_log()?;
1398        let mut emitted = 0;
1399        for stage_id in stage_ids {
1400            let attestation = lex_vcs::Attestation::new(
1401                stage_id,
1402                Some(op_id.clone()),
1403                None,
1404                lex_vcs::AttestationKind::Trace {
1405                    run_id: run_id.into(),
1406                    root_target: root_target.into(),
1407                },
1408                result.clone(),
1409                producer.clone(),
1410                None,
1411            );
1412            attlog.put(&attestation)?;
1413            emitted += 1;
1414        }
1415        Ok(emitted)
1416    }
1417
1418    /// Walk `ops_since(branch_head, base)` and emit per-stage
1419    /// `Trace` attestations for each new op, linking them to the
1420    /// run that produced them (#257). Used by `lex run --trace`
1421    /// after the VM exits: snapshot `base = branch_head` before
1422    /// the run, then call this with the post-run head.
1423    ///
1424    /// `base = None` means "every op currently reachable from the
1425    /// branch head" — generally not what you want for a single
1426    /// run; pass the pre-run head.
1427    ///
1428    /// Returns the total number of attestations emitted across
1429    /// every new op. Zero is the common case (the run committed no
1430    /// ops).
1431    ///
1432    /// Idempotent on the per-op level via [`Self::record_op_trace`].
1433    pub fn record_run_committed_ops_since(
1434        &self,
1435        run_id: &str,
1436        root_target: &str,
1437        branch: &str,
1438        base: Option<&lex_vcs::OpId>,
1439        result: lex_vcs::AttestationResult,
1440        producer: lex_vcs::ProducerDescriptor,
1441    ) -> Result<usize, StoreError> {
1442        let head = match self.get_branch(branch)?.and_then(|b| b.head_op) {
1443            Some(h) => h,
1444            None => return Ok(0),
1445        };
1446        let log = lex_vcs::OpLog::open(self.root())?;
1447        let new_ops = log.ops_since(&head, base)?;
1448        let mut total = 0;
1449        for rec in new_ops {
1450            total += self.record_op_trace(
1451                run_id,
1452                root_target,
1453                &rec.op_id,
1454                result.clone(),
1455                producer.clone(),
1456            )?;
1457        }
1458        Ok(total)
1459    }
1460
1461    /// Apply a typed `ReplaceMatchArm` transform (#280) and emit a
1462    /// `OperationKind::ReplaceMatchArm` op that records the
1463    /// semantic shape of the edit, not just the byte effect.
1464    ///
1465    /// Steps:
1466    ///   1. Load the source stage's canonical bytes (delta-aware).
1467    ///   2. Run [`lex_ast::replace_match_arm`] to produce the new
1468    ///      `Stage`. Pure function, no I/O.
1469    ///   3. Publish the new stage. Idempotent on the
1470    ///      content-addressed `to_stage_id`.
1471    ///   4. Assemble the candidate program (every active stage on
1472    ///      the branch, with the rewritten one swapped in) and call
1473    ///      [`Self::apply_operation_checked`] — re-typechecks and
1474    ///      runs every existing gate (TypeCheck attestation,
1475    ///      required_attestations, producer-block walk-back).
1476    ///
1477    /// Failure modes:
1478    ///   * [`StoreError::TransformError`] — transform didn't apply.
1479    ///     The branch is unchanged; no stage published.
1480    ///   * [`StoreError::TypeError`] — transform produced an
1481    ///     ill-typed program. The new stage is on disk (idempotent
1482    ///     on its content hash) but the branch is unchanged. Same
1483    ///     "publish without advance" semantics as #245.
1484    ///   * Everything else from `apply_operation_checked`.
1485    pub fn apply_replace_match_arm(
1486        &self,
1487        branch: &str,
1488        from_stage_id: &str,
1489        match_node: &lex_ast::NodeId,
1490        arm_index: usize,
1491        new_body: lex_ast::CExpr,
1492    ) -> Result<lex_vcs::OpId, StoreError> {
1493        let from_stage = self.get_ast(from_stage_id)?;
1494        let new_stage = lex_ast::replace_match_arm(&from_stage, match_node, arm_index, new_body)
1495            .map_err(StoreError::TransformError)?;
1496        let sig = lex_ast::sig_id(&from_stage).ok_or(StoreError::CannotPublishImport)?;
1497        let to_stage_id = self.publish(&new_stage)?;
1498        if to_stage_id == from_stage_id {
1499            // No-op transform — the new body was structurally
1500            // identical to the old. Refuse rather than advancing
1501            // the branch with an empty edit.
1502            return Err(StoreError::InvalidTransition(format!(
1503                "replace_match_arm produced the same stage_id `{from_stage_id}`"
1504            )));
1505        }
1506
1507        // Assemble the candidate program: every active stage on
1508        // the branch, with `from_stage_id` swapped for `new_stage`.
1509        let head = self.branch_head(branch)?;
1510        let mut candidate: Vec<lex_ast::Stage> = Vec::with_capacity(head.len());
1511        for (other_sig, other_stage_id) in &head {
1512            if other_sig == &sig {
1513                candidate.push(new_stage.clone());
1514            } else {
1515                candidate.push(self.get_ast(other_stage_id)?);
1516            }
1517        }
1518        // If the source sig isn't on the current branch head, the
1519        // transform is operating on a stage that hasn't been added
1520        // yet — refuse rather than risking a candidate program
1521        // that doesn't reflect the branch's actual state.
1522        if !head.contains_key(&sig) {
1523            return Err(StoreError::InvalidTransition(format!(
1524                "sig `{sig}` not on branch `{branch}`'s head"
1525            )));
1526        }
1527
1528        // #247: budget delta captured for `lex op log --budget-drift`.
1529        let from_budget = budget_of_stage(&from_stage);
1530        let to_budget = budget_of_stage(&new_stage);
1531
1532        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
1533        let kind = lex_vcs::OperationKind::ReplaceMatchArm {
1534            sig_id: sig.clone(),
1535            from_stage_id: from_stage_id.to_string(),
1536            to_stage_id: to_stage_id.clone(),
1537            match_node: match_node.as_str().to_string(),
1538            arm_index,
1539            from_budget,
1540            to_budget,
1541        };
1542        let transition = lex_vcs::StageTransition::Replace {
1543            sig_id: sig.clone(),
1544            from: from_stage_id.to_string(),
1545            to: to_stage_id.clone(),
1546        };
1547        let op = lex_vcs::Operation::new(kind, head_now.into_iter().collect::<Vec<_>>());
1548        self.apply_operation_checked(branch, op, transition, &candidate)
1549    }
1550
1551    /// Apply a typed `RenameLocal` transform (#280) — rename a
1552    /// `let`-bound local within a fn body and emit a matching
1553    /// `OperationKind::RenameLocal`. Same end-to-end shape as
1554    /// [`Self::apply_replace_match_arm`]; see that method for the
1555    /// failure-mode taxonomy.
1556    pub fn apply_rename_local(
1557        &self,
1558        branch: &str,
1559        from_stage_id: &str,
1560        let_node: &lex_ast::NodeId,
1561        new_name: &str,
1562    ) -> Result<lex_vcs::OpId, StoreError> {
1563        let from_stage = self.get_ast(from_stage_id)?;
1564        // Read the old name before running the transform, so the
1565        // op log records the rename target rather than just the
1566        // new value.
1567        let old_name = read_let_name(&from_stage, let_node).map_err(StoreError::TransformError)?;
1568        let new_stage = lex_ast::rename_local(&from_stage, let_node, new_name)
1569            .map_err(StoreError::TransformError)?;
1570        let sig = lex_ast::sig_id(&from_stage).ok_or(StoreError::CannotPublishImport)?;
1571        let to_stage_id = self.publish(&new_stage)?;
1572        if to_stage_id == from_stage_id {
1573            return Err(StoreError::InvalidTransition(format!(
1574                "rename_local produced the same stage_id `{from_stage_id}`"
1575            )));
1576        }
1577        let head = self.branch_head(branch)?;
1578        let mut candidate: Vec<lex_ast::Stage> = Vec::with_capacity(head.len());
1579        for (other_sig, other_stage_id) in &head {
1580            if other_sig == &sig {
1581                candidate.push(new_stage.clone());
1582            } else {
1583                candidate.push(self.get_ast(other_stage_id)?);
1584            }
1585        }
1586        if !head.contains_key(&sig) {
1587            return Err(StoreError::InvalidTransition(format!(
1588                "sig `{sig}` not on branch `{branch}`'s head"
1589            )));
1590        }
1591        let from_budget = budget_of_stage(&from_stage);
1592        let to_budget = budget_of_stage(&new_stage);
1593        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
1594        let kind = lex_vcs::OperationKind::RenameLocal {
1595            sig_id: sig.clone(),
1596            from_stage_id: from_stage_id.to_string(),
1597            to_stage_id: to_stage_id.clone(),
1598            let_node: let_node.as_str().to_string(),
1599            old_name,
1600            new_name: new_name.to_string(),
1601            from_budget,
1602            to_budget,
1603        };
1604        let transition = lex_vcs::StageTransition::Replace {
1605            sig_id: sig.clone(),
1606            from: from_stage_id.to_string(),
1607            to: to_stage_id.clone(),
1608        };
1609        let op = lex_vcs::Operation::new(kind, head_now.into_iter().collect::<Vec<_>>());
1610        self.apply_operation_checked(branch, op, transition, &candidate)
1611    }
1612
1613    /// Apply a typed `InlineLet` transform (#280) — eliminate a
1614    /// `let x := v; body` by substituting `v` for every unshadowed
1615    /// `x` in `body`, then replacing the `Let` node with the
1616    /// substituted body. Same end-to-end shape as
1617    /// [`Self::apply_replace_match_arm`].
1618    pub fn apply_inline_let(
1619        &self,
1620        branch: &str,
1621        from_stage_id: &str,
1622        let_node: &lex_ast::NodeId,
1623    ) -> Result<lex_vcs::OpId, StoreError> {
1624        let from_stage = self.get_ast(from_stage_id)?;
1625        let binding_name =
1626            read_let_name(&from_stage, let_node).map_err(StoreError::TransformError)?;
1627        let new_stage =
1628            lex_ast::inline_let(&from_stage, let_node).map_err(StoreError::TransformError)?;
1629        let sig = lex_ast::sig_id(&from_stage).ok_or(StoreError::CannotPublishImport)?;
1630        let to_stage_id = self.publish(&new_stage)?;
1631        if to_stage_id == from_stage_id {
1632            return Err(StoreError::InvalidTransition(format!(
1633                "inline_let produced the same stage_id `{from_stage_id}`"
1634            )));
1635        }
1636        let head = self.branch_head(branch)?;
1637        let mut candidate: Vec<lex_ast::Stage> = Vec::with_capacity(head.len());
1638        for (other_sig, other_stage_id) in &head {
1639            if other_sig == &sig {
1640                candidate.push(new_stage.clone());
1641            } else {
1642                candidate.push(self.get_ast(other_stage_id)?);
1643            }
1644        }
1645        if !head.contains_key(&sig) {
1646            return Err(StoreError::InvalidTransition(format!(
1647                "sig `{sig}` not on branch `{branch}`'s head"
1648            )));
1649        }
1650        let from_budget = budget_of_stage(&from_stage);
1651        let to_budget = budget_of_stage(&new_stage);
1652        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
1653        let kind = lex_vcs::OperationKind::InlineLet {
1654            sig_id: sig.clone(),
1655            from_stage_id: from_stage_id.to_string(),
1656            to_stage_id: to_stage_id.clone(),
1657            let_node: let_node.as_str().to_string(),
1658            binding_name,
1659            from_budget,
1660            to_budget,
1661        };
1662        let transition = lex_vcs::StageTransition::Replace {
1663            sig_id: sig.clone(),
1664            from: from_stage_id.to_string(),
1665            to: to_stage_id.clone(),
1666        };
1667        let op = lex_vcs::Operation::new(kind, head_now.into_iter().collect::<Vec<_>>());
1668        self.apply_operation_checked(branch, op, transition, &candidate)
1669    }
1670
1671    /// Apply a typed `ExtractFunction` transform (#280 slice 4) —
1672    /// extract a sub-expression of `from_stage_id`'s body into a
1673    /// new top-level fn defined by `spec`, and emit two ops tied
1674    /// together by a shared synthetic Intent so `lex op log
1675    /// --intent <id>` groups them.
1676    ///
1677    /// The two ops:
1678    ///   1. `AddFunction { sig_id: <new_fn_sig>, stage_id: <new_fn_stage> }`
1679    ///   2. `ModifyBody { sig_id: <source_sig>, from_stage_id, to_stage_id: <modified> }`
1680    ///
1681    /// The shared Intent's prompt is structured (`extract_function:
1682    /// <new_fn_name>` plus the source identity) so downstream
1683    /// tooling can recover the typed-transform shape from the
1684    /// op-log + intent-log join.
1685    ///
1686    /// Returns `(add_fn_op_id, modify_body_op_id)`.
1687    pub fn apply_extract_function(
1688        &self,
1689        branch: &str,
1690        from_stage_id: &str,
1691        expr_node: &lex_ast::NodeId,
1692        spec: lex_ast::ExtractFnSpec,
1693    ) -> Result<(lex_vcs::OpId, lex_vcs::OpId), StoreError> {
1694        let from_stage = self.get_ast(from_stage_id)?;
1695        let new_fn_name = spec.name.clone();
1696        let (modified_stage, new_fn_stage) =
1697            lex_ast::extract_function(&from_stage, expr_node, spec)
1698                .map_err(StoreError::TransformError)?;
1699
1700        let source_sig = lex_ast::sig_id(&from_stage).ok_or(StoreError::CannotPublishImport)?;
1701        let new_fn_sig = lex_ast::sig_id(&new_fn_stage).ok_or(StoreError::CannotPublishImport)?;
1702        if source_sig == new_fn_sig {
1703            return Err(StoreError::InvalidTransition(format!(
1704                "extract_function produced a sig matching the source `{source_sig}`"
1705            )));
1706        }
1707        let new_fn_stage_id = self.publish(&new_fn_stage)?;
1708        let modified_stage_id = self.publish(&modified_stage)?;
1709        if modified_stage_id == from_stage_id {
1710            return Err(StoreError::InvalidTransition(format!(
1711                "extract_function produced the same stage_id `{from_stage_id}` for the source"
1712            )));
1713        }
1714
1715        let head = self.branch_head(branch)?;
1716        if !head.contains_key(&source_sig) {
1717            return Err(StoreError::InvalidTransition(format!(
1718                "sig `{source_sig}` not on branch `{branch}`'s head"
1719            )));
1720        }
1721
1722        // Synthesize an Intent linking the two ops. The session_id
1723        // / model fields here are not load-bearing — they exist to
1724        // make the IntentId content-addressed; downstream tooling
1725        // reads `prompt` to reconstruct the typed-transform shape.
1726        let intent = lex_vcs::Intent::new(
1727            format!(
1728                "[lex.transform.extract_function]\nnew_fn={new_fn_name}\nsource_sig={source_sig}\nfrom_stage={from_stage_id}\nexpr_node={node}",
1729                node = expr_node.as_str(),
1730            ),
1731            "lex-store::apply_extract_function",
1732            lex_vcs::ModelDescriptor {
1733                provider: "lex-store".into(),
1734                name: env!("CARGO_PKG_VERSION").into(),
1735                version: None,
1736            },
1737            None,
1738        );
1739        let intent_id = intent.intent_id.clone();
1740        lex_vcs::IntentLog::open(self.root())?.put(&intent)?;
1741
1742        // Step 1 — emit the AddFunction op for the new fn. Build
1743        // the candidate program by appending the new fn to every
1744        // stage on the current branch head.
1745        let new_fn_effects: std::collections::BTreeSet<String> = match &new_fn_stage {
1746            lex_ast::Stage::FnDecl(fd) => fd.effects.iter().map(|e| e.name.clone()).collect(),
1747            _ => Default::default(),
1748        };
1749        let new_fn_budget = budget_of_stage(&new_fn_stage);
1750        let mut candidate_with_new_fn: Vec<lex_ast::Stage> = Vec::with_capacity(head.len() + 1);
1751        for stage_id in head.values() {
1752            candidate_with_new_fn.push(self.get_ast(stage_id)?);
1753        }
1754        candidate_with_new_fn.push(new_fn_stage.clone());
1755        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
1756        let add_op = lex_vcs::Operation::new(
1757            lex_vcs::OperationKind::AddFunction {
1758                sig_id: new_fn_sig.clone(),
1759                stage_id: new_fn_stage_id.clone(),
1760                effects: new_fn_effects,
1761                budget_cost: new_fn_budget,
1762            },
1763            head_now.into_iter().collect::<Vec<_>>(),
1764        )
1765        .with_intent(intent_id.clone());
1766        let add_transition = lex_vcs::StageTransition::Create {
1767            sig_id: new_fn_sig.clone(),
1768            stage_id: new_fn_stage_id.clone(),
1769        };
1770        let add_op_id =
1771            self.apply_operation_checked(branch, add_op, add_transition, &candidate_with_new_fn)?;
1772
1773        // Step 2 — emit the ModifyBody op for the source. Build
1774        // the candidate program by replacing the source's stage
1775        // with `modified_stage` and keeping the new fn alongside.
1776        let from_budget = budget_of_stage(&from_stage);
1777        let to_budget = budget_of_stage(&modified_stage);
1778        let mut candidate_with_modified: Vec<lex_ast::Stage> = Vec::with_capacity(head.len() + 1);
1779        for (other_sig, other_stage_id) in &head {
1780            if other_sig == &source_sig {
1781                candidate_with_modified.push(modified_stage.clone());
1782            } else {
1783                candidate_with_modified.push(self.get_ast(other_stage_id)?);
1784            }
1785        }
1786        candidate_with_modified.push(new_fn_stage.clone());
1787        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
1788        let modify_op = lex_vcs::Operation::new(
1789            lex_vcs::OperationKind::ModifyBody {
1790                sig_id: source_sig.clone(),
1791                from_stage_id: from_stage_id.to_string(),
1792                to_stage_id: modified_stage_id.clone(),
1793                from_budget,
1794                to_budget,
1795            },
1796            head_now.into_iter().collect::<Vec<_>>(),
1797        )
1798        .with_intent(intent_id);
1799        let modify_transition = lex_vcs::StageTransition::Replace {
1800            sig_id: source_sig,
1801            from: from_stage_id.to_string(),
1802            to: modified_stage_id,
1803        };
1804        let modify_op_id = self.apply_operation_checked(
1805            branch,
1806            modify_op,
1807            modify_transition,
1808            &candidate_with_modified,
1809        )?;
1810
1811        Ok((add_op_id, modify_op_id))
1812    }
1813
1814    /// Propose a stage for `sig_id` without advancing the branch
1815    /// head (#294). Multiple agents can call this concurrently
1816    /// for the same sig — every call lands a fresh `Candidate`
1817    /// op chained off the current head_op. The branch head stays
1818    /// where it was; a later [`Self::promote_candidate`] picks
1819    /// the winner.
1820    ///
1821    /// The caller is responsible for typechecking `new_stage`
1822    /// against whatever program context they consider valid —
1823    /// `propose_candidate` doesn't run the gate. Type errors
1824    /// surface at promotion time, where the candidate is
1825    /// composed back into a candidate program via the standard
1826    /// `apply_operation_checked` path.
1827    ///
1828    /// The stage is published (idempotent on content hash). The
1829    /// `intent_id` is required so downstream consumers can
1830    /// distinguish proposals by author.
1831    pub fn propose_candidate(
1832        &self,
1833        branch: &str,
1834        new_stage: &lex_ast::Stage,
1835        intent_id: &lex_vcs::IntentId,
1836    ) -> Result<lex_vcs::OpId, StoreError> {
1837        let sig = lex_ast::sig_id(new_stage).ok_or(StoreError::CannotPublishImport)?;
1838        let stage_id = self.publish(new_stage)?;
1839        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
1840        let op = lex_vcs::Operation::new(
1841            lex_vcs::OperationKind::Candidate {
1842                sig_id: sig,
1843                stage_id,
1844            },
1845            head_now.into_iter().collect::<Vec<_>>(),
1846        )
1847        .with_intent(intent_id.clone());
1848        let transition = lex_vcs::StageTransition::ImportOnly;
1849        self.apply_operation(branch, op, transition)
1850    }
1851
1852    /// List every live `Candidate` op for `sig_id` — i.e. those
1853    /// not yet referenced by any `Promote` op (either as the
1854    /// winner or in the `supersedes` set). Used by `lex stage
1855    /// candidates`. Results are sorted by op_id for
1856    /// reproducibility.
1857    pub fn list_candidates(&self, sig_id: &str) -> Result<Vec<CandidateInfo>, StoreError> {
1858        let log = lex_vcs::OpLog::open(self.root())?;
1859        let all = log.list_all()?;
1860        // Collect the set of candidate op_ids referenced by any
1861        // Promote for this sig. Those candidates are no longer
1862        // live.
1863        let mut referenced: std::collections::BTreeSet<lex_vcs::OpId> = Default::default();
1864        for rec in &all {
1865            if let lex_vcs::OperationKind::Promote {
1866                sig_id: s,
1867                winner_candidate,
1868                supersedes,
1869                ..
1870            } = &rec.op.kind
1871            {
1872                if s != sig_id {
1873                    continue;
1874                }
1875                referenced.insert(winner_candidate.clone());
1876                for sup in supersedes {
1877                    referenced.insert(sup.clone());
1878                }
1879            }
1880        }
1881        let mut out: Vec<CandidateInfo> = Vec::new();
1882        for rec in all {
1883            let lex_vcs::OperationKind::Candidate {
1884                sig_id: s,
1885                stage_id,
1886            } = &rec.op.kind
1887            else {
1888                continue;
1889            };
1890            if s != sig_id {
1891                continue;
1892            }
1893            if referenced.contains(&rec.op_id) {
1894                continue;
1895            }
1896            out.push(CandidateInfo {
1897                op_id: rec.op_id.clone(),
1898                stage_id: stage_id.clone(),
1899                intent_id: rec.op.intent_id.clone(),
1900            });
1901        }
1902        out.sort_by(|a, b| a.op_id.cmp(&b.op_id));
1903        Ok(out)
1904    }
1905
1906    /// Promote a previously-landed `Candidate` op as the new
1907    /// branch head for its sig (#294). Emits a `Promote` op
1908    /// listing every other live `Candidate` for the same sig
1909    /// in its `supersedes` field. After this lands,
1910    /// [`Self::list_candidates`] returns an empty set for the
1911    /// sig.
1912    ///
1913    /// Re-typechecks the candidate program (winner stage + the
1914    /// rest of the branch) through `apply_operation_checked`, so
1915    /// a candidate that doesn't compose with the current branch
1916    /// state surfaces as `StoreError::TypeError`.
1917    pub fn promote_candidate(
1918        &self,
1919        branch: &str,
1920        candidate_op_id: &lex_vcs::OpId,
1921    ) -> Result<lex_vcs::OpId, StoreError> {
1922        let log = lex_vcs::OpLog::open(self.root())?;
1923        let candidate_rec = log
1924            .get(candidate_op_id)?
1925            .ok_or_else(|| StoreError::UnknownOp(candidate_op_id.clone()))?;
1926        let (sig, winner_stage_id) = match &candidate_rec.op.kind {
1927            lex_vcs::OperationKind::Candidate { sig_id, stage_id } => {
1928                (sig_id.clone(), stage_id.clone())
1929            }
1930            other => {
1931                return Err(StoreError::InvalidTransition(format!(
1932                    "op `{candidate_op_id}` is a `{:?}`, not a Candidate",
1933                    other
1934                )))
1935            }
1936        };
1937
1938        // Gather every OTHER live candidate for this sig — the
1939        // ones this Promote will supersede.
1940        let live = self.list_candidates(&sig)?;
1941        let mut supersedes: Vec<lex_vcs::OpId> = live
1942            .iter()
1943            .filter(|c| &c.op_id != candidate_op_id)
1944            .map(|c| c.op_id.clone())
1945            .collect();
1946        supersedes.sort();
1947
1948        // Assemble candidate program: winner stage in place of
1949        // the sig's current head (if any), plus every other sig
1950        // unchanged.
1951        let head = self.branch_head(branch)?;
1952        let winner_stage = self.get_ast(&winner_stage_id)?;
1953        let mut candidate_program: Vec<lex_ast::Stage> = Vec::with_capacity(head.len() + 1);
1954        let mut found = false;
1955        for (other_sig, other_stage_id) in &head {
1956            if other_sig == &sig {
1957                candidate_program.push(winner_stage.clone());
1958                found = true;
1959            } else {
1960                candidate_program.push(self.get_ast(other_stage_id)?);
1961            }
1962        }
1963        if !found {
1964            // Sig doesn't have a head yet — append the winner
1965            // stage to make it a Create.
1966            candidate_program.push(winner_stage.clone());
1967        }
1968        let from_stage_id = head.get(&sig).cloned();
1969        // Budget delta from old head to winner — same shape as
1970        // ModifyBody.
1971        let from_budget = from_stage_id
1972            .as_deref()
1973            .and_then(|s| self.get_ast(s).ok())
1974            .and_then(|s| budget_of_stage(&s));
1975        let to_budget = budget_of_stage(&winner_stage);
1976
1977        let head_now = self.get_branch(branch)?.and_then(|b| b.head_op);
1978        let op = lex_vcs::Operation::new(
1979            lex_vcs::OperationKind::Promote {
1980                sig_id: sig.clone(),
1981                winner_candidate: candidate_op_id.clone(),
1982                winner_stage_id: winner_stage_id.clone(),
1983                supersedes,
1984                from_stage_id: from_stage_id.clone(),
1985                from_budget,
1986                to_budget,
1987            },
1988            head_now.into_iter().collect::<Vec<_>>(),
1989        );
1990        let transition = match &from_stage_id {
1991            Some(from) => lex_vcs::StageTransition::Replace {
1992                sig_id: sig,
1993                from: from.clone(),
1994                to: winner_stage_id,
1995            },
1996            None => lex_vcs::StageTransition::Create {
1997                sig_id: sig,
1998                stage_id: winner_stage_id,
1999            },
2000        };
2001        self.apply_operation_checked(branch, op, transition, &candidate_program)
2002    }
2003
2004    /// `set_branch_head_op` for the durability story on the branch
2005    /// file itself.
2006    pub fn apply_operation(
2007        &self,
2008        branch: &str,
2009        op: lex_vcs::Operation,
2010        transition: lex_vcs::StageTransition,
2011    ) -> Result<lex_vcs::OpId, StoreError> {
2012        let attestable = attestable_stage_ids(&transition);
2013        let op_effects = op_declared_effects(&op.kind);
2014        self.cas_retry_advance(branch, op, transition, |new_head| {
2015            self.run_required_attestations_gate(branch, &new_head.op_id, &attestable, &op_effects)
2016        })
2017    }
2018
2019    /// CAS retry loop for #262. Single-parent ops are rebuilt on
2020    /// each iteration with the current branch head as parent;
2021    /// the per-iteration callback runs the gate (and TypeCheck
2022    /// emission, for the checked path) between persist and CAS.
2023    /// Merge ops (with 2 parents already set) skip the rebuild —
2024    /// their parents are caller-supplied and meaningful — and get
2025    /// a single attempt; on CAS failure they surface `Contention`.
2026    fn cas_retry_advance<F>(
2027        &self,
2028        branch: &str,
2029        op: lex_vcs::Operation,
2030        transition: lex_vcs::StageTransition,
2031        mut between_persist_and_cas: F,
2032    ) -> Result<lex_vcs::OpId, StoreError>
2033    where
2034        F: FnMut(&lex_vcs::NewHead) -> Result<(), StoreError>,
2035    {
2036        // 32 retries handles up to ~32 concurrent writers racing on
2037        // the same branch tip. Beyond that, surfacing `Contention`
2038        // is the right signal — clients should back off or batch.
2039        const MAX_ATTEMPTS: u32 = 32;
2040        // Single-parent ops can be rebuilt on retry; merge ops
2041        // can't (their two parents are meaningful, supplied by the
2042        // merge engine). For merges, single attempt: if CAS
2043        // fails, surface Contention.
2044        let is_rebuildable = op.parents.len() <= 1;
2045        let kind = op.kind.clone();
2046        let intent_id = op.intent_id.clone();
2047
2048        let mut last_io_err: Option<StoreError> = None;
2049        let mut current_op = op;
2050        let current_transition = transition;
2051        // Only rebuild on retries — attempt 1 honors the caller's
2052        // exact op so a user-supplied bogus parent (parents =
2053        // ["someone-else"]) surfaces as `StaleParent` instead of
2054        // being silently corrected.
2055        //
2056        // Exception (#262 follow-up): an op with `parents = []`
2057        // means "I don't care; chain off whatever the current
2058        // head is." Under concurrent apply, attempt 1 can read
2059        // `head_op = Some(opA)` after a sibling writer landed,
2060        // and the persist's parent check fails StaleParent
2061        // unprompted. Rebuild attempt 1 for the empty-parents
2062        // case so the legitimate-race path retries cleanly.
2063        let mut rebuilt_already = false;
2064        for attempt in 1..=MAX_ATTEMPTS {
2065            // Read the current head BEFORE we persist — this is
2066            // the value we'll compare against in the CAS.
2067            let parent = self.get_branch(branch)?.and_then(|b| b.head_op);
2068
2069            // Rebuild the op against the current head, but only
2070            // on retries (not the caller's first attempt) and
2071            // only for single-parent operations. Multi-parent
2072            // (merge) ops are passed through unchanged.
2073            //
2074            // Empty-parents ops also rebuild on attempt 1 (see
2075            // the exception note above) so concurrent apply
2076            // doesn't false-positive on StaleParent.
2077            let should_rebuild = is_rebuildable
2078                && (rebuilt_already || (current_op.parents.is_empty() && parent.is_some()));
2079            if should_rebuild {
2080                current_op = lex_vcs::Operation {
2081                    kind: kind.clone(),
2082                    parents: parent.iter().cloned().collect(),
2083                    intent_id: intent_id.clone(),
2084                };
2085            }
2086
2087            // Persist (idempotent). On `StaleParent` from a retry
2088            // attempt (where we already rebuilt), the head changed
2089            // between our `get_branch` and this `lex_vcs::apply`
2090            // — race; rebuild and continue. On `StaleParent` from
2091            // attempt 1 (caller's input), propagate.
2092            let new_head = match self.persist_op_only_with_parent(
2093                branch,
2094                parent.as_ref(),
2095                current_op.clone(),
2096                current_transition.clone(),
2097            ) {
2098                Ok(nh) => nh,
2099                Err(StoreError::Apply(lex_vcs::ApplyError::StaleParent { .. }))
2100                    if is_rebuildable && rebuilt_already =>
2101                {
2102                    rebuilt_already = true;
2103                    continue;
2104                }
2105                Err(e) => return Err(e),
2106            };
2107
2108            // Run the caller's between-persist-and-cas hook
2109            // (TypeCheck emission + gate). If this fails, the op
2110            // record is durable but orphaned — same semantics as
2111            // pre-#262.
2112            between_persist_and_cas(&new_head)?;
2113
2114            // CAS the branch head. On success: done. On mismatch:
2115            // someone advanced in parallel; retry.
2116            match self.set_branch_head_op_cas(branch, parent, new_head.op_id.clone()) {
2117                Ok(()) => return Ok(new_head.op_id),
2118                Err(crate::branches::CasFailed::Mismatch { .. }) if is_rebuildable => {
2119                    // Try again with the new head as parent.
2120                    rebuilt_already = true;
2121                    continue;
2122                }
2123                Err(crate::branches::CasFailed::Mismatch { .. }) => {
2124                    // Merge op: surface immediately — we can't
2125                    // rebuild without rerunning the merge engine.
2126                    let _ = attempt;
2127                    return Err(StoreError::Contention {
2128                        branch: branch.into(),
2129                        attempts: 1,
2130                    });
2131                }
2132                Err(crate::branches::CasFailed::UnknownBranch(b)) => {
2133                    return Err(StoreError::UnknownBranch(b));
2134                }
2135                Err(crate::branches::CasFailed::Io(e)) => {
2136                    last_io_err = Some(StoreError::Io(std::io::Error::other(e)));
2137                    continue;
2138                }
2139            }
2140        }
2141        // Retries exhausted. Prefer surfacing the most recent IO
2142        // error if we hit one; otherwise it's pure CAS contention.
2143        match last_io_err {
2144            Some(e) => Err(e),
2145            None => Err(StoreError::Contention {
2146                branch: branch.into(),
2147                attempts: MAX_ATTEMPTS,
2148            }),
2149        }
2150    }
2151
2152    /// Persist an op against an explicitly-supplied parent. Used
2153    /// by the CAS retry loop in `cas_retry_advance` so the
2154    /// `lex_vcs::apply` parent check matches what we read at the
2155    /// top of the loop iteration (avoids a TOCTOU race against
2156    /// `persist_op_only`'s second read).
2157    fn persist_op_only_with_parent(
2158        &self,
2159        branch: &str,
2160        parent: Option<&lex_vcs::OpId>,
2161        op: lex_vcs::Operation,
2162        transition: lex_vcs::StageTransition,
2163    ) -> Result<lex_vcs::NewHead, StoreError> {
2164        if branch != DEFAULT_BRANCH && self.get_branch(branch)?.is_none() {
2165            return Err(StoreError::UnknownBranch(branch.into()));
2166        }
2167        let log = lex_vcs::OpLog::open(self.root())?;
2168        lex_vcs::apply(&log, parent, op, transition).map_err(|e| match e {
2169            lex_vcs::ApplyError::Persist(io) => StoreError::Io(io),
2170            other => StoreError::Apply(other),
2171        })
2172    }
2173
2174    /// Run the `required_attestations` gate (#245) and the
2175    /// retroactive producer-block gate (#248) over a single op
2176    /// against the store's `policy.json` and attestation log.
2177    ///
2178    /// Failure modes (in order):
2179    ///
2180    /// 1. Producer-block first: if any attestation on the op's
2181    ///    stage is from a quarantined tool, refuse with
2182    ///    `ProducerBlocked` (#248). Surfaces *before* the
2183    ///    required-attestations gate so a clearly-malicious record
2184    ///    isn't masked by a missing-Spec error.
2185    /// 2. Required-attestations next: if any required attestation
2186    ///    kind is missing, refuse with `BranchAdvanceBlocked`
2187    ///    (#245).
2188    ///
2189    /// Loads the policy / attestation log lazily; with no policy
2190    /// file and no `ProducerBlock` attestations the gate is a no-op
2191    /// (default-permissive — matches pre-#245 stores).
2192    fn run_required_attestations_gate(
2193        &self,
2194        branch: &str,
2195        op_id: &lex_vcs::OpId,
2196        stage_ids: &[String],
2197        op_effects: &std::collections::BTreeSet<String>,
2198    ) -> Result<(), StoreError> {
2199        // Build the candidate slice for the new op. Ops with no
2200        // attestable stage (imports, empty merges) get a single
2201        // `None`-stage tuple; both gates skip those.
2202        let new_op_candidate: Vec<(
2203            lex_vcs::OpId,
2204            Option<String>,
2205            std::collections::BTreeSet<String>,
2206        )> = if stage_ids.is_empty() {
2207            vec![(op_id.clone(), None, op_effects.clone())]
2208        } else {
2209            stage_ids
2210                .iter()
2211                .map(|sid| (op_id.clone(), Some(sid.clone()), op_effects.clone()))
2212                .collect()
2213        };
2214        let attest_log = self.attestation_log()?;
2215
2216        // #248 + #256: producer-block gate, walk-back style.
2217        //
2218        // The naive #248 gate only checked the new op's stage. That
2219        // missed contamination on ancestors — once `lex attest
2220        // retro-block` lands, every previously-gated op stays in
2221        // the chain even though its attestations are now from a
2222        // quarantined producer.
2223        //
2224        // #256 fixes this by walking the chain from `head_op` back
2225        // to `last_gate_checkpoint` (or genesis when the checkpoint
2226        // is invalidated), collecting each ancestor's attestable
2227        // stages, and running `check_producer_block` on the
2228        // combined set. After a successful advance,
2229        // `set_branch_head_op` moves the checkpoint to the new
2230        // head (steady-state O(new ops) per advance).
2231        let walk_back_candidate = self.collect_ancestor_candidates(branch)?;
2232        let mut producer_block_candidate = walk_back_candidate;
2233        producer_block_candidate.extend(new_op_candidate.iter().cloned());
2234        crate::policy::check_producer_block(&attest_log, &producer_block_candidate)
2235            .map_err(StoreError::ProducerBlocked)?;
2236
2237        // #245: required-attestations gate. Forward-going only —
2238        // only the new op is checked. Walking back makes no sense
2239        // here: the policy is "this advance must carry these
2240        // attestations," not "every prior op must have."
2241        let policy = match crate::policy::load(self.root())? {
2242            Some(p) if !p.required_attestations.is_empty() => p,
2243            _ => return Ok(()),
2244        };
2245        let waivers =
2246            crate::policy::check_required_attestations(&attest_log, &new_op_candidate, &policy)
2247                .map_err(StoreError::BranchAdvanceBlocked)?;
2248        // #293: emit one `TrustWaived` attestation per waiver so
2249        // the audit trail records every skip. Idempotent on
2250        // attestation_id (content-addressed dedup) — re-running
2251        // the gate with the same state writes the same files.
2252        for w in waivers {
2253            let att = lex_vcs::Attestation::new(
2254                w.stage_id,
2255                Some(op_id.clone()),
2256                None,
2257                lex_vcs::AttestationKind::TrustWaived {
2258                    producer: w.producer,
2259                    score_thousandths: w.score_thousandths,
2260                    threshold_thousandths: w.threshold_thousandths,
2261                    kind_tag: w.kind_tag,
2262                },
2263                lex_vcs::AttestationResult::Passed,
2264                trust_waived_producer(),
2265                None,
2266            );
2267            attest_log.put(&att)?;
2268        }
2269        Ok(())
2270    }
2271
2272    /// Walk the branch from `head_op` back to `last_gate_checkpoint`
2273    /// (exclusive) and return the `(op_id, stage_id, op_effects)`
2274    /// tuples for every attestable stage touched by an ancestor
2275    /// (#256). Empty when the branch is fresh, when the checkpoint
2276    /// equals the head, or when the head is None.
2277    fn collect_ancestor_candidates(&self, branch: &str) -> Result<Vec<GateCandidate>, StoreError> {
2278        let b = match self.get_branch(branch)? {
2279            Some(b) => b,
2280            None => return Ok(Vec::new()),
2281        };
2282        let Some(head) = b.head_op else {
2283            return Ok(Vec::new());
2284        };
2285        if Some(&head) == b.last_gate_checkpoint.as_ref() {
2286            // Steady-state common case: previous advance left the
2287            // checkpoint at head. Nothing to re-walk.
2288            return Ok(Vec::new());
2289        }
2290
2291        let log = lex_vcs::OpLog::open(self.root())?;
2292        let walk = log.walk_back(&head, None)?;
2293        let stop_at = b.last_gate_checkpoint.clone();
2294        let mut out = Vec::new();
2295        for rec in walk {
2296            if Some(&rec.op_id) == stop_at.as_ref() {
2297                break;
2298            }
2299            let stages = attestable_stage_ids(&rec.produces);
2300            let effects = op_declared_effects(&rec.op.kind);
2301            if stages.is_empty() {
2302                out.push((rec.op_id.clone(), None, effects));
2303            } else {
2304                for sid in stages {
2305                    out.push((rec.op_id.clone(), Some(sid), effects.clone()));
2306                }
2307            }
2308        }
2309        Ok(out)
2310    }
2311}
2312
2313fn stage_name(stage: &Stage) -> &str {
2314    match stage {
2315        Stage::FnDecl(fd) => &fd.name,
2316        Stage::TypeDecl(td) => &td.name,
2317        Stage::Import(i) => &i.alias,
2318    }
2319}
2320
2321fn stage_for_kind<'a>(
2322    kind: &lex_vcs::OperationKind,
2323    stages: &'a [lex_ast::Stage],
2324) -> Option<&'a lex_ast::Stage> {
2325    use lex_vcs::OperationKind::*;
2326    let target_sig = match kind {
2327        AddFunction { sig_id, .. }
2328        | ModifyBody { sig_id, .. }
2329        | ChangeEffectSig { sig_id, .. }
2330        | AddType { sig_id, .. }
2331        | ModifyType { sig_id, .. } => Some(sig_id.clone()),
2332        RenameSymbol { to, .. } => Some(to.clone()),
2333        _ => None,
2334    };
2335    let target_sig = target_sig?;
2336    stages
2337        .iter()
2338        .find(|s| sig_id(s).as_deref() == Some(target_sig.as_str()))
2339}
2340
2341fn transition_for_kind(kind: &lex_vcs::OperationKind) -> lex_vcs::StageTransition {
2342    use lex_vcs::OperationKind::*;
2343    use lex_vcs::StageTransition;
2344    match kind {
2345        AddFunction {
2346            sig_id, stage_id, ..
2347        }
2348        | AddType { sig_id, stage_id } => StageTransition::Create {
2349            sig_id: sig_id.clone(),
2350            stage_id: stage_id.clone(),
2351        },
2352        RemoveFunction {
2353            sig_id,
2354            last_stage_id,
2355        }
2356        | RemoveType {
2357            sig_id,
2358            last_stage_id,
2359        } => StageTransition::Remove {
2360            sig_id: sig_id.clone(),
2361            last: last_stage_id.clone(),
2362        },
2363        ModifyBody {
2364            sig_id,
2365            from_stage_id,
2366            to_stage_id,
2367            ..
2368        }
2369        | ChangeEffectSig {
2370            sig_id,
2371            from_stage_id,
2372            to_stage_id,
2373            ..
2374        }
2375        | ModifyType {
2376            sig_id,
2377            from_stage_id,
2378            to_stage_id,
2379        }
2380        | ReplaceMatchArm {
2381            sig_id,
2382            from_stage_id,
2383            to_stage_id,
2384            ..
2385        }
2386        | RenameLocal {
2387            sig_id,
2388            from_stage_id,
2389            to_stage_id,
2390            ..
2391        }
2392        | InlineLet {
2393            sig_id,
2394            from_stage_id,
2395            to_stage_id,
2396            ..
2397        } => StageTransition::Replace {
2398            sig_id: sig_id.clone(),
2399            from: from_stage_id.clone(),
2400            to: to_stage_id.clone(),
2401        },
2402        RenameSymbol {
2403            from,
2404            to,
2405            body_stage_id,
2406        } => StageTransition::Rename {
2407            from: from.clone(),
2408            to: to.clone(),
2409            body_stage_id: body_stage_id.clone(),
2410        },
2411        AddImport { .. } | RemoveImport { .. } => StageTransition::ImportOnly,
2412        Merge { .. } => StageTransition::Merge {
2413            entries: Default::default(),
2414        },
2415        // #294: a Candidate proposes a stage without advancing
2416        // the branch. ImportOnly keeps the branch head untouched
2417        // — the stage IS published on disk (Store::propose_candidate
2418        // calls publish before apply), but no head delta lands.
2419        Candidate { .. } => StageTransition::ImportOnly,
2420        // A Promote advances the head exactly like ModifyBody
2421        // (or Create when the sig had no head). The winner
2422        // stage is the new branch state for that sig.
2423        Promote {
2424            sig_id,
2425            winner_stage_id,
2426            from_stage_id,
2427            ..
2428        } => match from_stage_id {
2429            Some(from) => StageTransition::Replace {
2430                sig_id: sig_id.clone(),
2431                from: from.clone(),
2432                to: winner_stage_id.clone(),
2433            },
2434            None => StageTransition::Create {
2435                sig_id: sig_id.clone(),
2436                stage_id: winner_stage_id.clone(),
2437            },
2438        },
2439    }
2440}
2441
2442/// Producer identity for TypeCheck attestations emitted by the
2443/// store-write gate. Pinned to this crate's name + version so an
2444/// attestation produced by a different `lex-store` revision is
2445/// distinguishable (content-hashed `produced_by`).
2446fn typecheck_producer() -> lex_vcs::ProducerDescriptor {
2447    lex_vcs::ProducerDescriptor {
2448        tool: "lex-store".into(),
2449        version: env!("CARGO_PKG_VERSION").into(),
2450        model: None,
2451    }
2452}
2453
2454/// Producer identity for `RepairHint` attestations emitted by
2455/// `apply_operation_checked` on TypeError (#281). Distinct tool
2456/// name from `typecheck_producer` so consumers can filter the
2457/// activity feed for repair hints without scanning kinds.
2458fn repair_hint_producer() -> lex_vcs::ProducerDescriptor {
2459    lex_vcs::ProducerDescriptor {
2460        tool: "lex-store::repair_hint".into(),
2461        version: env!("CARGO_PKG_VERSION").into(),
2462        model: None,
2463    }
2464}
2465
2466/// Producer identity for `TrustWaived` attestations emitted by
2467/// the `required_attestations` gate on a trust-driven waiver
2468/// (#293). Distinct from `typecheck_producer` and `repair_hint`
2469/// so the audit trail clearly shows "the gate let this advance
2470/// through because trust > threshold."
2471fn trust_waived_producer() -> lex_vcs::ProducerDescriptor {
2472    lex_vcs::ProducerDescriptor {
2473        tool: "lex-store::trust_waived".into(),
2474        version: env!("CARGO_PKG_VERSION").into(),
2475        model: None,
2476    }
2477}
2478
2479/// Producer identity for `ProducerTrust` attestations emitted by
2480/// [`Store::recompute_producer_trust`]. The score-derivation
2481/// recompute is its own machine-emittable kind, distinct from
2482/// the gate-side `TrustWaived` emit (#293).
2483fn producer_trust_producer() -> lex_vcs::ProducerDescriptor {
2484    lex_vcs::ProducerDescriptor {
2485        tool: "lex-store::producer_trust".into(),
2486        version: env!("CARGO_PKG_VERSION").into(),
2487        model: None,
2488    }
2489}
2490
2491/// The set of stage_ids a transition introduces. These are the
2492/// stages a successful TypeCheck pass attests *about* — the new
2493/// head produced by Create/Replace, the renamed body, or the per-
2494/// sig resolution of a Merge. Removes and ImportOnly produce no
2495/// attestable stage; the program typechecks but no specific stage
2496/// is the subject of the claim.
2497/// One row of input to the producer-block / required-attestations
2498/// gates: `(op_id, stage_id, op_effects)`. The `stage_id` is
2499/// `None` for ops that don't touch a stage (imports, empty
2500/// merges) — the gate skips those.
2501type GateCandidate = (
2502    lex_vcs::OpId,
2503    Option<String>,
2504    std::collections::BTreeSet<String>,
2505);
2506
2507/// Effect set declared *by the operation itself* (#245). Used by
2508/// the `required_attestations` gate's `EffectsIntersect` clause.
2509///
2510/// Only `AddFunction` and `ChangeEffectSig` carry an effect set in
2511/// their op payload; for everything else this returns the empty
2512/// set, which means `EffectsIntersect` rules don't fire on those
2513/// ops. `Always` rules continue to fire regardless. A future
2514/// improvement is to extract effects from the candidate `Stage`
2515/// for `ModifyBody` ops, but the typed-effects-on-ops path (#247)
2516/// is the cleaner solution and lands separately.
2517fn op_declared_effects(kind: &lex_vcs::OperationKind) -> std::collections::BTreeSet<String> {
2518    use lex_vcs::OperationKind::*;
2519    match kind {
2520        AddFunction { effects, .. } => effects.clone(),
2521        ChangeEffectSig { to_effects, .. } => to_effects.clone(),
2522        _ => std::collections::BTreeSet::new(),
2523    }
2524}
2525
2526fn attestable_stage_ids(transition: &lex_vcs::StageTransition) -> Vec<String> {
2527    use lex_vcs::StageTransition::*;
2528    match transition {
2529        Create { stage_id, .. } => vec![stage_id.clone()],
2530        Replace { to, .. } => vec![to.clone()],
2531        Rename { body_stage_id, .. } => vec![body_stage_id.clone()],
2532        Merge { entries } => entries.values().filter_map(|opt| opt.clone()).collect(),
2533        Remove { .. } | ImportOnly => Vec::new(),
2534    }
2535}
2536
2537fn write_canonical_json<T: Serialize>(path: &Path, value: &T) -> Result<(), StoreError> {
2538    let v = serde_json::to_value(value)?;
2539    let s = lex_ast::canon_json::to_canonical_string(&v);
2540    if let Some(parent) = path.parent() {
2541        fs::create_dir_all(parent)?;
2542    }
2543    fs::write(path, s)?;
2544    Ok(())
2545}
2546
2547/// Read the `name` of the `Let` expression at `let_node` inside
2548/// `stage`'s body. Used by [`Store::apply_rename_local`] to record
2549/// the rename source. Returns the same `TransformError` shapes as
2550/// the transformer itself so callers see a consistent error
2551/// vocabulary.
2552fn read_let_name(
2553    stage: &Stage,
2554    let_node: &lex_ast::NodeId,
2555) -> Result<String, lex_ast::TransformError> {
2556    // The transformer is itself a pure function; ask it to perform
2557    // a rename to a sentinel value and read the resulting let's
2558    // original name from the output. Cheaper than duplicating the
2559    // node-walk here, and stays correct as the transform evolves.
2560    //
2561    // We use a sentinel that's invalid as a Lex identifier so even
2562    // if the rename somehow lands, downstream parsing would
2563    // surface it loudly. (The transform path discards the renamed
2564    // value — we only need the *original* name.)
2565    let probed = lex_ast::rename_local(stage, let_node, "__lex_rename_probe__")?;
2566    let Stage::FnDecl(fd) = probed else {
2567        return Err(lex_ast::TransformError::NonFnTarget {
2568            stage_kind: "non-FnDecl",
2569        });
2570    };
2571    // Walk back to the probed let to read its old name from the
2572    // *original* stage — the probed stage's let has already been
2573    // renamed.
2574    let Stage::FnDecl(orig_fd) = stage else {
2575        return Err(lex_ast::TransformError::NonFnTarget {
2576            stage_kind: "non-FnDecl",
2577        });
2578    };
2579    // Path-based lookup matches the transformer's navigation.
2580    let path = parse_let_node_path(let_node.as_str())?;
2581    if path.is_empty() {
2582        return Err(lex_ast::TransformError::NotALet {
2583            at: let_node.as_str().into(),
2584            found_kind: "stage_root",
2585        });
2586    }
2587    if path[0] != orig_fd.params.len() + 1 {
2588        return Err(lex_ast::TransformError::UnknownNode {
2589            at: let_node.as_str().into(),
2590        });
2591    }
2592    let inner = &path[1..];
2593    let target = navigate_to_let(&orig_fd.body, inner, let_node.as_str())?;
2594    let _ = fd; // probed stage discarded
2595    Ok(target.to_string())
2596}
2597
2598fn parse_let_node_path(id: &str) -> Result<Vec<usize>, lex_ast::TransformError> {
2599    let s = id
2600        .strip_prefix("n_")
2601        .ok_or_else(|| lex_ast::TransformError::BadNodeId(id.into()))?;
2602    let mut parts = s.split('.');
2603    let head = parts
2604        .next()
2605        .ok_or_else(|| lex_ast::TransformError::BadNodeId(id.into()))?;
2606    if head != "0" {
2607        return Err(lex_ast::TransformError::BadNodeId(id.into()));
2608    }
2609    let mut out = Vec::new();
2610    for p in parts {
2611        out.push(
2612            p.parse::<usize>()
2613                .map_err(|_| lex_ast::TransformError::BadNodeId(id.into()))?,
2614        );
2615    }
2616    Ok(out)
2617}
2618
2619fn navigate_to_let<'a>(
2620    root: &'a lex_ast::CExpr,
2621    path: &[usize],
2622    at: &str,
2623) -> Result<&'a str, lex_ast::TransformError> {
2624    use lex_ast::CExpr::*;
2625    let mut current = root;
2626    for &idx in path {
2627        current = match current {
2628            Call { callee, args } => {
2629                if idx == 0 {
2630                    callee
2631                } else {
2632                    args.get(idx - 1)
2633                        .ok_or_else(|| lex_ast::TransformError::UnknownNode { at: at.into() })?
2634                }
2635            }
2636            Let { value, body, .. } => match idx {
2637                0 => value,
2638                1 => body,
2639                _ => return Err(lex_ast::TransformError::UnknownNode { at: at.into() }),
2640            },
2641            Match { scrutinee, arms } => {
2642                if idx == 0 {
2643                    scrutinee
2644                } else {
2645                    let arm_off = idx - 1;
2646                    if arm_off % 2 != 1 {
2647                        return Err(lex_ast::TransformError::UnknownNode { at: at.into() });
2648                    }
2649                    let arm_index = arm_off / 2;
2650                    &arms
2651                        .get(arm_index)
2652                        .ok_or_else(|| lex_ast::TransformError::UnknownNode { at: at.into() })?
2653                        .body
2654                }
2655            }
2656            Block { statements, result } => {
2657                if idx < statements.len() {
2658                    &statements[idx]
2659                } else if idx == statements.len() {
2660                    result
2661                } else {
2662                    return Err(lex_ast::TransformError::UnknownNode { at: at.into() });
2663                }
2664            }
2665            Constructor { args, .. }
2666            | TupleLit { items: args, .. }
2667            | ListLit { items: args, .. } => args
2668                .get(idx)
2669                .ok_or_else(|| lex_ast::TransformError::UnknownNode { at: at.into() })?,
2670            RecordLit { fields } => {
2671                &fields
2672                    .get(idx)
2673                    .ok_or_else(|| lex_ast::TransformError::UnknownNode { at: at.into() })?
2674                    .value
2675            }
2676            FieldAccess { value, .. } if idx == 0 => value,
2677            Lambda { body, .. } if idx == 0 => body,
2678            BinOp { lhs, rhs, .. } => match idx {
2679                0 => lhs,
2680                1 => rhs,
2681                _ => return Err(lex_ast::TransformError::UnknownNode { at: at.into() }),
2682            },
2683            UnaryOp { expr, .. } if idx == 0 => expr,
2684            Return { value } if idx == 0 => value,
2685            _ => return Err(lex_ast::TransformError::UnknownNode { at: at.into() }),
2686        };
2687    }
2688    let Let { name, .. } = current else {
2689        return Err(lex_ast::TransformError::NotALet {
2690            at: at.into(),
2691            found_kind: lex_cexpr_kind(current),
2692        });
2693    };
2694    Ok(name)
2695}
2696
2697fn lex_cexpr_kind(e: &lex_ast::CExpr) -> &'static str {
2698    use lex_ast::CExpr::*;
2699    match e {
2700        Literal { .. } => "Literal",
2701        Var { .. } => "Var",
2702        Call { .. } => "Call",
2703        Let { .. } => "Let",
2704        Match { .. } => "Match",
2705        Block { .. } => "Block",
2706        Constructor { .. } => "Constructor",
2707        RecordLit { .. } => "RecordLit",
2708        TupleLit { .. } => "TupleLit",
2709        ListLit { .. } => "ListLit",
2710        FieldAccess { .. } => "FieldAccess",
2711        Lambda { .. } => "Lambda",
2712        BinOp { .. } => "BinOp",
2713        UnaryOp { .. } => "UnaryOp",
2714        Return { .. } => "Return",
2715    }
2716}
2717
2718/// Extract the declared `[budget(N)]` integer from a stage's
2719/// effect set, if any (#280 + #247). Returns `None` for stages
2720/// that aren't `FnDecl` or don't carry a budget effect — same
2721/// shape as `lex_vcs::budget_from_effects`.
2722fn budget_of_stage(stage: &Stage) -> Option<u64> {
2723    let fd = match stage {
2724        Stage::FnDecl(fd) => fd,
2725        _ => return None,
2726    };
2727    let mut min_cost: Option<u64> = None;
2728    for eff in &fd.effects {
2729        if eff.name != "budget" {
2730            continue;
2731        }
2732        if let Some(lex_ast::EffectArg::Int { value }) = &eff.arg {
2733            let n = *value as u64;
2734            min_cost = Some(min_cost.map(|c| c.min(n)).unwrap_or(n));
2735        }
2736    }
2737    min_cost
2738}
2739
2740/// Serialize a stage to its canonical-JSON byte form. Used by
2741/// `publish_signed` for delta encoding (#261 slice 3) — both the
2742/// "compute the diff" path and the "write a full snapshot"
2743/// fallback need exactly the same bytes.
2744fn canonical_bytes(stage: &Stage) -> Result<Vec<u8>, StoreError> {
2745    let v = serde_json::to_value(stage)?;
2746    Ok(lex_ast::canon_json::to_canonical_string(&v).into_bytes())
2747}
2748
2749#[allow(dead_code)]
2750fn read_json<T: DeserializeOwned>(path: &Path) -> Result<T, StoreError> {
2751    let bytes = fs::read(path)?;
2752    Ok(serde_json::from_slice(&bytes)?)
2753}