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