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