Skip to main content

memstead_base/
pipeline_store.rs

1//! File-adapter persistence for the pipeline configs.
2//!
3//! The **live** store is one record kind: a v2 [`Binding`] per pipeline at
4//! `<root>/.memstead/projections/<mem>/<name>.json`, read version-gated by
5//! [`load_pipeline_configs`]. The record's identity is `(mem, name)` derived
6//! from the file path; `name` is the file stem.
7//!
8//! Everything else here is **migrate-local** — the retired store layouts the
9//! `memstead projection migrate` legs parse and clean up, never served live:
10//!
11//! - `<root>/.memstead/mediums/<mem>/<name>.json` — legacy [`Medium`]
12//! - `<root>/.memstead/facets/<mem>/<name>.json` — legacy [`Facet`]
13//! - version-less `projections/…` files — legacy gen-2 [`Projection`]
14//! - `<root>/.memstead/ingests/<name>.json` — flat legacy [`LegacyIngest`]
15//!
16//! The loader's job is load + validate + expose read-only: a malformed config
17//! surfaces a typed [`StoreError::Parse`] naming the offending file (the
18//! early-validation value), rather than being silently skipped; a pre-v2
19//! record surfaces [`StoreError::LegacyProjectionStore`] naming the migrate
20//! command.
21
22use std::path::{Path, PathBuf};
23
24use serde::Serialize;
25use serde::de::DeserializeOwned;
26
27use serde::Deserialize;
28
29use crate::binding::Binding;
30use crate::pipeline::{Facet, IngestTrigger, Medium, Projection};
31use crate::workspace_store::{StoreError, WORKSPACE_STORE_DIR};
32
33/// The legacy (gen-2) flat **ingest** record — the schedule primitive the
34/// binding model retired (its `mode`/`trigger`/`batch_size`/`deny_paths`/
35/// `post_actions` collapsed into a binding's `operations.build` block).
36///
37/// This shape is **migrate-local**: it is parsed only by the legacy path —
38/// [`load_legacy_pipeline_configs`], `projection migrate` (which reads it to
39/// merge each ingest into its projection), the referential-integrity edit
40/// layer, and the gen-1 root-folder converter. No live surface constructs or
41/// runs it; the live loader speaks [`Binding`]. Kept `pub(crate)` and minimal
42/// so the retired `Ingest`/`IngestMode` machinery no longer lives on any public
43/// or live surface.
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45pub(crate) struct LegacyIngest {
46    /// The projection (by name, `"<mem>/<name>"`) this ingest ran.
47    pub projection: String,
48    /// Discovery / refinement / one-shot — including the deleted `refinement`
49    /// value, parsed so `projection migrate` can *detect and refuse* it (D1).
50    pub mode: LegacyIngestMode,
51    /// Loop / manual / on-event.
52    pub trigger: IngestTrigger,
53    /// How many artifacts a single run processed.
54    pub batch_size: u32,
55    /// Paths excluded for this ingest's runs, on top of facet scope.
56    #[serde(default)]
57    pub deny_paths: Vec<String>,
58    /// Free-form post-run actions (e.g. a one-shot `archive_source` flag).
59    #[serde(default, skip_serializing_if = "Option::is_none")]
60    pub post_actions: Option<serde_json::Value>,
61}
62
63/// The legacy ingest mode, including the deleted `refinement` value. Parsed so
64/// `projection migrate` can detect and refuse it (D1) — never carried forward
65/// into a binding.
66#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
67#[serde(rename_all = "kebab-case")]
68pub(crate) enum LegacyIngestMode {
69    /// Build out new coverage.
70    Discovery,
71    /// Improve existing coverage — the deleted refinement-as-writer mode.
72    Refinement,
73    /// A single bounded pass.
74    OneShot,
75}
76
77/// Per-primitive subdirectory names under the workspace store.
78pub const MEDIUMS_DIR: &str = "mediums";
79/// See [`MEDIUMS_DIR`].
80pub const FACETS_DIR: &str = "facets";
81/// See [`MEDIUMS_DIR`].
82pub const PROJECTIONS_DIR: &str = "projections";
83/// See [`MEDIUMS_DIR`].
84pub const INGESTS_DIR: &str = "ingests";
85
86/// A per-mem pipeline record paired with the mem and name (file stem) that
87/// identify it on disk.
88#[derive(Debug, Clone, PartialEq, Serialize)]
89pub struct MemPipelineRecord<T> {
90    /// The mem subdirectory this record lives under.
91    pub mem: String,
92    /// The record's name — the file stem (e.g. `source-tree`).
93    pub name: String,
94    /// The parsed config.
95    pub config: T,
96}
97
98/// A flat (non-per-mem) pipeline record — used for ingests.
99#[derive(Debug, Clone, PartialEq, Serialize)]
100pub struct PipelineRecord<T> {
101    /// The record's name — the file stem (e.g. `macos-graph`).
102    pub name: String,
103    /// The parsed config.
104    pub config: T,
105}
106
107/// Every pipeline config in a workspace store, in the four-primitive shape.
108///
109/// This is the **legacy** (gen-2) shape — produced only by
110/// [`load_legacy_pipeline_configs`] and consumed only by the
111/// `projection migrate` conversion legs. The live loader
112/// ([`load_pipeline_configs`]) returns [`BindingConfigs`] instead.
113#[derive(Debug, Default, Clone, PartialEq, Serialize)]
114pub struct PipelineConfigs {
115    /// Per-mem mediums.
116    pub mediums: Vec<MemPipelineRecord<Medium>>,
117    /// Per-mem facets.
118    pub facets: Vec<MemPipelineRecord<Facet>>,
119    /// Per-mem projections.
120    pub projections: Vec<MemPipelineRecord<Projection>>,
121    /// Flat legacy ingests (crate-local — the migrate/edit path reads them; no
122    /// live or external surface does).
123    pub(crate) ingests: Vec<PipelineRecord<LegacyIngest>>,
124}
125
126/// Every pipeline config in a workspace store, in the **v2 single-record**
127/// shape: one [`Binding`] per pipeline, nothing else. This is what the live
128/// loader [`load_pipeline_configs`] returns and the brief / selection /
129/// cursor paths consume. Canonical id is `<mem>/<stem>`.
130#[derive(Debug, Default, Clone, PartialEq, Serialize)]
131pub struct BindingConfigs {
132    /// Per-mem v2 bindings under the `projections/<mem>/<name>.json` tier.
133    pub bindings: Vec<MemPipelineRecord<Binding>>,
134    /// Bindings whose stored file failed the version gate or parse —
135    /// quarantined instead of failing the whole load (degrade, never
136    /// disappear; agent-trust plan 04). A quarantined binding serves
137    /// no operations: resolution sites refuse typed with the entry's
138    /// reason (whose message names `memstead projection migrate` for
139    /// the legacy generations). Mems and healthy bindings serve
140    /// normally.
141    #[serde(default, skip_serializing_if = "Vec::is_empty")]
142    pub quarantined: Vec<QuarantinedBinding>,
143}
144
145/// One quarantined binding: the store file that failed the v2
146/// version gate or parse, with its typed reason.
147#[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
148pub struct QuarantinedBinding {
149    /// Destination mem (the `projections/<mem>/` tier).
150    pub mem: String,
151    /// Binding name (`<name>.json`).
152    pub name: String,
153    /// Offending file path.
154    pub path: String,
155    /// Typed reason code (`PROJECTION_STORE_LEGACY`,
156    /// `UNKNOWN_BINDING_VERSION`, `WORKSPACE_STORE_PARSE`).
157    pub reason_code: String,
158    /// Full reason message — the legacy generations' message names
159    /// `memstead projection migrate`.
160    pub reason_message: String,
161    /// Reconstruction payload for [`load_pipeline_configs_strict`]:
162    /// the declared version on an `UNKNOWN_BINDING_VERSION` entry.
163    #[serde(default, skip_serializing_if = "Option::is_none")]
164    pub unknown_version: Option<i64>,
165    /// Reconstruction payload: the serde message on a
166    /// `WORKSPACE_STORE_PARSE` entry.
167    #[serde(default, skip_serializing_if = "Option::is_none")]
168    pub parse_message: Option<String>,
169}
170
171/// The `<root>/.memstead/<primitive>` directory for a given primitive.
172fn primitive_dir(workspace_root: &Path, primitive: &str) -> PathBuf {
173    workspace_root.join(WORKSPACE_STORE_DIR).join(primitive)
174}
175
176/// Refuse a `mem`/`name` value that is not a single, plain path
177/// component: separators, traversal segments, drive/stream colons, NULs,
178/// and empty values would let a caller-supplied name write or delete
179/// outside the workspace's own metadata directory. Validated here — the
180/// one place every mutation's path is built — so no surface above
181/// (CLI, UniFFI, engine) can bypass it.
182fn validate_component(kind: &str, value: &str) -> Result<(), StoreError> {
183    let invalid = value.is_empty()
184        || value == "."
185        || value == ".."
186        || value.contains('/')
187        || value.contains('\\')
188        || value.contains(':')
189        || value.contains('\0');
190    if invalid {
191        return Err(StoreError::Other(format!(
192            "invalid {kind} '{}': must be a single path component \
193             (no separators, traversal segments, ':' or NUL)",
194            value.escape_default()
195        )));
196    }
197    Ok(())
198}
199
200/// File path of a per-mem record: `<root>/.memstead/<primitive>/<mem>/<name>.json`.
201fn mem_scoped_path(
202    workspace_root: &Path,
203    primitive: &str,
204    mem: &str,
205    name: &str,
206) -> Result<PathBuf, StoreError> {
207    validate_component("mem", mem)?;
208    validate_component("name", name)?;
209    Ok(primitive_dir(workspace_root, primitive)
210        .join(mem)
211        .join(format!("{name}.json")))
212}
213
214/// File path of a flat (non-per-mem) record: `<root>/.memstead/<primitive>/<name>.json`.
215fn flat_path(workspace_root: &Path, primitive: &str, name: &str) -> Result<PathBuf, StoreError> {
216    validate_component("name", name)?;
217    Ok(primitive_dir(workspace_root, primitive).join(format!("{name}.json")))
218}
219
220/// Remove the file at `path`, mapping IO failures (including a missing
221/// file) to a typed [`StoreError::Io`] naming the path. Dumb counterpart
222/// to [`write_json`] — referential-integrity / existence checks belong to
223/// the calling layer, matching the write-is-upsert / load-validates split.
224fn remove_file(path: &Path) -> Result<(), StoreError> {
225    std::fs::remove_file(path).map_err(|e| StoreError::Io {
226        path: path.to_path_buf(),
227        source: e,
228    })
229}
230
231/// Rename the record file `from` → `to`. Refuses to clobber an existing
232/// target (silent overwrite would lose a distinct record); that guard is
233/// the one non-dumb concession here because the failure mode is data loss.
234/// A missing source surfaces as [`StoreError::Io`]. Reference rewriting in
235/// dependent primitives is the calling layer's job.
236fn rename_file(from: &Path, to: &Path) -> Result<(), StoreError> {
237    if to.exists() {
238        return Err(StoreError::Other(format!(
239            "rename target already exists: {}",
240            to.display()
241        )));
242    }
243    std::fs::rename(from, to).map_err(|e| StoreError::Io {
244        path: from.to_path_buf(),
245        source: e,
246    })
247}
248
249/// Serialise `config` (pretty JSON) into `path`, creating parent directories.
250fn write_json<T: Serialize>(path: &Path, config: &T) -> Result<(), StoreError> {
251    if let Some(parent) = path.parent() {
252        std::fs::create_dir_all(parent).map_err(|e| StoreError::Io {
253            path: parent.to_path_buf(),
254            source: e,
255        })?;
256    }
257    let bytes = serde_json::to_vec_pretty(config).map_err(|e| StoreError::Parse {
258        path: path.to_path_buf(),
259        message: e.to_string(),
260    })?;
261    std::fs::write(path, bytes).map_err(|e| StoreError::Io {
262        path: path.to_path_buf(),
263        source: e,
264    })
265}
266
267/// Load every `<primitive>/<mem>/<name>.json` under the store, parsed.
268/// Absent primitive directory → empty (a workspace may declare no pipelines).
269/// A malformed file surfaces a typed parse error naming the path.
270fn load_mem_scoped<T: DeserializeOwned>(
271    workspace_root: &Path,
272    primitive: &str,
273) -> Result<Vec<MemPipelineRecord<T>>, StoreError> {
274    let dir = primitive_dir(workspace_root, primitive);
275    let mut out: Vec<MemPipelineRecord<T>> = Vec::new();
276    let mem_dirs = match std::fs::read_dir(&dir) {
277        Ok(rd) => rd,
278        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out),
279        Err(e) => {
280            return Err(StoreError::Io {
281                path: dir,
282                source: e,
283            });
284        }
285    };
286    for mem_entry in mem_dirs.flatten() {
287        let mem_path = mem_entry.path();
288        if !mem_path.is_dir() {
289            continue;
290        }
291        let mem = mem_entry.file_name().to_string_lossy().into_owned();
292        let files = match std::fs::read_dir(&mem_path) {
293            Ok(rd) => rd,
294            Err(e) => {
295                return Err(StoreError::Io {
296                    path: mem_path,
297                    source: e,
298                });
299            }
300        };
301        for file in files.flatten() {
302            let path = file.path();
303            if path.extension().and_then(|e| e.to_str()) != Some("json") {
304                continue;
305            }
306            let Some(name) = path.file_stem().map(|s| s.to_string_lossy().into_owned()) else {
307                continue;
308            };
309            let config = read_json::<T>(&path)?;
310            out.push(MemPipelineRecord {
311                mem: mem.clone(),
312                name,
313                config,
314            });
315        }
316    }
317    // Deterministic order so callers (and tests) see a stable enumeration.
318    out.sort_by(|a, b| (a.mem.as_str(), a.name.as_str()).cmp(&(b.mem.as_str(), b.name.as_str())));
319    Ok(out)
320}
321
322/// Load every `ingests/<name>.json` (flat), parsed.
323fn load_flat<T: DeserializeOwned>(
324    workspace_root: &Path,
325    primitive: &str,
326) -> Result<Vec<PipelineRecord<T>>, StoreError> {
327    let dir = primitive_dir(workspace_root, primitive);
328    let mut out: Vec<PipelineRecord<T>> = Vec::new();
329    let files = match std::fs::read_dir(&dir) {
330        Ok(rd) => rd,
331        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out),
332        Err(e) => {
333            return Err(StoreError::Io {
334                path: dir,
335                source: e,
336            });
337        }
338    };
339    for file in files.flatten() {
340        let path = file.path();
341        if path.extension().and_then(|e| e.to_str()) != Some("json") {
342            continue;
343        }
344        let Some(name) = path.file_stem().map(|s| s.to_string_lossy().into_owned()) else {
345            continue;
346        };
347        let config = read_json::<T>(&path)?;
348        out.push(PipelineRecord { name, config });
349    }
350    out.sort_by(|a, b| a.name.cmp(&b.name));
351    Ok(out)
352}
353
354/// Read + parse one JSON file into `T`, mapping IO/parse failures to typed
355/// [`StoreError`]s naming the path.
356fn read_json<T: DeserializeOwned>(path: &Path) -> Result<T, StoreError> {
357    let bytes = std::fs::read(path).map_err(|e| StoreError::Io {
358        path: path.to_path_buf(),
359        source: e,
360    })?;
361    serde_json::from_slice(&bytes).map_err(|e| StoreError::Parse {
362        path: path.to_path_buf(),
363        message: e.to_string(),
364    })
365}
366
367/// Write a medium to `<root>/.memstead/mediums/<mem>/<name>.json`.
368pub fn write_medium(
369    workspace_root: &Path,
370    mem: &str,
371    name: &str,
372    medium: &Medium,
373) -> Result<(), StoreError> {
374    write_json(
375        &mem_scoped_path(workspace_root, MEDIUMS_DIR, mem, name)?,
376        medium,
377    )
378}
379
380/// Write a facet to `<root>/.memstead/facets/<mem>/<name>.json`.
381pub fn write_facet(
382    workspace_root: &Path,
383    mem: &str,
384    name: &str,
385    facet: &Facet,
386) -> Result<(), StoreError> {
387    write_json(
388        &mem_scoped_path(workspace_root, FACETS_DIR, mem, name)?,
389        facet,
390    )
391}
392
393/// Write a projection to `<root>/.memstead/projections/<mem>/<name>.json`.
394pub fn write_projection(
395    workspace_root: &Path,
396    mem: &str,
397    name: &str,
398    projection: &Projection,
399) -> Result<(), StoreError> {
400    write_json(
401        &mem_scoped_path(workspace_root, PROJECTIONS_DIR, mem, name)?,
402        projection,
403    )
404}
405
406/// Write a legacy ingest to `<root>/.memstead/ingests/<name>.json` (flat).
407/// Crate-local dumb file op — used by the gen-1 root-folder converter and the
408/// projection-rename repoint; the live path writes bindings, not ingests.
409pub(crate) fn write_ingest(
410    workspace_root: &Path,
411    name: &str,
412    ingest: &LegacyIngest,
413) -> Result<(), StoreError> {
414    write_json(&flat_path(workspace_root, INGESTS_DIR, name)?, ingest)
415}
416
417/// Write a v2 binding to `<root>/.memstead/projections/<mem>/<name>.json`.
418///
419/// A binding occupies the *same* per-mem projections tier and file identity
420/// its predecessors did (stem-identity preserved), so this overwrites a
421/// prior-generation projection file in place when migrating a workspace.
422pub fn write_binding(
423    workspace_root: &Path,
424    mem: &str,
425    name: &str,
426    binding: &Binding,
427) -> Result<(), StoreError> {
428    write_json(
429        &mem_scoped_path(workspace_root, PROJECTIONS_DIR, mem, name)?,
430        binding,
431    )
432}
433
434/// Read the v2 binding at `<root>/.memstead/projections/<mem>/<name>.json`.
435///
436/// The read counterpart of [`write_binding`] — reads the *same* per-mem
437/// projections tier and file identity, parsed as a [`Binding`]. A missing
438/// file surfaces [`StoreError::Io`] (kind `NotFound`); a file present but not
439/// a v2 binding (e.g. a not-yet-migrated store) surfaces
440/// [`StoreError::Parse`]. Callers wanting a friendly "no such binding"
441/// message pre-check existence and keep the two apart.
442pub fn read_binding(workspace_root: &Path, mem: &str, name: &str) -> Result<Binding, StoreError> {
443    read_json(&mem_scoped_path(
444        workspace_root,
445        PROJECTIONS_DIR,
446        mem,
447        name,
448    )?)
449}
450
451/// Delete a medium file. Missing → [`StoreError::Io`]; callers that want a
452/// friendly "no such medium" pre-check existence via [`load_pipeline_configs`].
453pub fn delete_medium(workspace_root: &Path, mem: &str, name: &str) -> Result<(), StoreError> {
454    remove_file(&mem_scoped_path(workspace_root, MEDIUMS_DIR, mem, name)?)
455}
456
457/// Delete a facet file. See [`delete_medium`] for missing-file semantics.
458pub fn delete_facet(workspace_root: &Path, mem: &str, name: &str) -> Result<(), StoreError> {
459    remove_file(&mem_scoped_path(workspace_root, FACETS_DIR, mem, name)?)
460}
461
462/// Delete a projection file. See [`delete_medium`] for missing-file semantics.
463pub fn delete_projection(workspace_root: &Path, mem: &str, name: &str) -> Result<(), StoreError> {
464    remove_file(&mem_scoped_path(
465        workspace_root,
466        PROJECTIONS_DIR,
467        mem,
468        name,
469    )?)
470}
471
472/// Delete an ingest file (flat). See [`delete_medium`] for missing-file semantics.
473pub fn delete_ingest(workspace_root: &Path, name: &str) -> Result<(), StoreError> {
474    remove_file(&flat_path(workspace_root, INGESTS_DIR, name)?)
475}
476
477// Rename is exposed only for the *nameless* records (projection, ingest),
478// whose identity is the file stem alone. Mediums and facets carry an embedded
479// `name` field that must equal the stem (facets reference mediums by name,
480// projections reference facets by name); a pure file move would leave that
481// field stale, so their rename lives in the `pipeline_edit` layer, which
482// rewrites the embedded name and dependent references together.
483
484/// Rename a projection within its mem (`old` → `new`, same `<mem>` tier).
485/// Refuses to clobber an existing target. A projection has no embedded name,
486/// so a file move is its whole rename; rewriting dependent ingest `projection`
487/// references is the calling layer's job.
488pub fn rename_projection(
489    workspace_root: &Path,
490    mem: &str,
491    old: &str,
492    new: &str,
493) -> Result<(), StoreError> {
494    rename_file(
495        &mem_scoped_path(workspace_root, PROJECTIONS_DIR, mem, old)?,
496        &mem_scoped_path(workspace_root, PROJECTIONS_DIR, mem, new)?,
497    )
498}
499
500/// Rename an ingest (flat). Refuses to clobber an existing target. An ingest
501/// has no embedded name and nothing references it, so a file move is the whole
502/// rename.
503pub fn rename_ingest(workspace_root: &Path, old: &str, new: &str) -> Result<(), StoreError> {
504    rename_file(
505        &flat_path(workspace_root, INGESTS_DIR, old)?,
506        &flat_path(workspace_root, INGESTS_DIR, new)?,
507    )
508}
509
510/// Load the **legacy** (gen-2) four-primitive store from the workspace
511/// (migrate-local). Absent directories resolve to empty; a malformed file
512/// surfaces a typed [`StoreError::Parse`]. This reader is the counterpart of
513/// the version-gated [`load_pipeline_configs`]: it deliberately reads the
514/// old `Projection` + flat-`Ingest` shape (parsing a versioned binding file
515/// lossily as a [`Projection`], which ignores `version`/`operations`) so the
516/// `projection migrate` legs can see prior generations. It performs **no**
517/// version gate — it is the escape hatch the gate points migrations at, and
518/// nothing live consumes it.
519pub fn load_legacy_pipeline_configs(workspace_root: &Path) -> Result<PipelineConfigs, StoreError> {
520    Ok(PipelineConfigs {
521        mediums: load_mem_scoped(workspace_root, MEDIUMS_DIR)?,
522        facets: load_mem_scoped(workspace_root, FACETS_DIR)?,
523        projections: load_mem_scoped(workspace_root, PROJECTIONS_DIR)?,
524        ingests: load_flat(workspace_root, INGESTS_DIR)?,
525    })
526}
527
528/// Load every `projections/<mem>/<name>.json` as a **v2 binding**,
529/// version-gated. Absent directory → empty. For each file:
530///
531/// - no `version` field (gen-2 projection) or `version: 1` (three-file-store
532///   binding) → [`StoreError::LegacyProjectionStore`] — a prior generation
533///   the loader never serves; the message names `memstead projection
534///   migrate`;
535/// - any other `version` but not `2` → [`StoreError::UnknownBindingVersion`];
536/// - `version: 2` → parsed as [`Binding`] (a malformed operations block etc.
537///   surfaces [`StoreError::Parse`] naming the file).
538///
539/// The gate refuses the whole load on the first offending file — a pre-v2
540/// workspace fails loudly (pointing at migrate) rather than loading half-served.
541fn load_bindings(
542    workspace_root: &Path,
543) -> Result<(Vec<MemPipelineRecord<Binding>>, Vec<QuarantinedBinding>), StoreError> {
544    let dir = primitive_dir(workspace_root, PROJECTIONS_DIR);
545    let mut out: Vec<MemPipelineRecord<Binding>> = Vec::new();
546    let mut quarantined: Vec<QuarantinedBinding> = Vec::new();
547    let mem_dirs = match std::fs::read_dir(&dir) {
548        Ok(rd) => rd,
549        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((out, quarantined)),
550        Err(e) => {
551            return Err(StoreError::Io {
552                path: dir,
553                source: e,
554            });
555        }
556    };
557    for mem_entry in mem_dirs.flatten() {
558        let mem_path = mem_entry.path();
559        if !mem_path.is_dir() {
560            continue;
561        }
562        let mem = mem_entry.file_name().to_string_lossy().into_owned();
563        let files = match std::fs::read_dir(&mem_path) {
564            Ok(rd) => rd,
565            Err(e) => {
566                return Err(StoreError::Io {
567                    path: mem_path,
568                    source: e,
569                });
570            }
571        };
572        for file in files.flatten() {
573            let path = file.path();
574            if path.extension().and_then(|e| e.to_str()) != Some("json") {
575                continue;
576            }
577            let Some(name) = path.file_stem().map(|s| s.to_string_lossy().into_owned()) else {
578                continue;
579            };
580            // Peek at `version` before committing to the Binding shape so a
581            // pre-v2 file yields the migrate-naming error rather than an
582            // opaque "missing field" parse error. Version-less (gen-2) and
583            // v1 (three-file store) are known prior generations → the
584            // migrate-naming refusal; anything else non-2 is unknown.
585            // A failing file QUARANTINES that binding instead of
586            // failing the whole load (degrade, never disappear;
587            // agent-trust plan 04) — the typed judgment is unchanged,
588            // the blast radius shrinks to the one binding.
589            let path_display = path.display().to_string();
590            let quarantine =
591                |q: &mut Vec<QuarantinedBinding>, mem: &str, name: String, e: StoreError| {
592                    let (unknown_version, parse_message) = match &e {
593                        StoreError::UnknownBindingVersion { version, .. } => (Some(*version), None),
594                        StoreError::Parse { message, .. } => (None, Some(message.clone())),
595                        _ => (None, None),
596                    };
597                    q.push(QuarantinedBinding {
598                        mem: mem.to_string(),
599                        name,
600                        path: path_display.clone(),
601                        reason_code: e.code().to_string(),
602                        reason_message: e.to_string(),
603                        unknown_version,
604                        parse_message,
605                    });
606                };
607            let value: serde_json::Value = match read_json(&path) {
608                Ok(v) => v,
609                Err(e) => {
610                    quarantine(&mut quarantined, &mem, name, e);
611                    continue;
612                }
613            };
614            let gate_err = match value.get("version") {
615                None => Some(StoreError::LegacyProjectionStore { path: path.clone() }),
616                Some(v) => match v.as_i64() {
617                    Some(1) => Some(StoreError::LegacyProjectionStore { path: path.clone() }),
618                    n if n != Some(i64::from(crate::binding::BINDING_VERSION)) => {
619                        Some(StoreError::UnknownBindingVersion {
620                            path: path.clone(),
621                            version: n.unwrap_or(-1),
622                        })
623                    }
624                    _ => None,
625                },
626            };
627            if let Some(e) = gate_err {
628                quarantine(&mut quarantined, &mem, name, e);
629                continue;
630            }
631            let config: Binding = match serde_json::from_value(value) {
632                Ok(c) => c,
633                Err(e) => {
634                    quarantine(
635                        &mut quarantined,
636                        &mem,
637                        name,
638                        StoreError::Parse {
639                            path: path.clone(),
640                            message: e.to_string(),
641                        },
642                    );
643                    continue;
644                }
645            };
646            out.push(MemPipelineRecord {
647                mem: mem.clone(),
648                name,
649                config,
650            });
651        }
652    }
653    out.sort_by(|a, b| (a.mem.as_str(), a.name.as_str()).cmp(&(b.mem.as_str(), b.name.as_str())));
654    quarantined
655        .sort_by(|a, b| (a.mem.as_str(), a.name.as_str()).cmp(&(b.mem.as_str(), b.name.as_str())));
656    Ok((out, quarantined))
657}
658
659/// Load the live **v2 single-record** store from the workspace:
660/// `projections/` read as version-gated v2 bindings via [`load_bindings`] — a
661/// pre-v2 file (version-less gen-2, or v1 three-file store) refuses with
662/// [`StoreError::LegacyProjectionStore`] naming `memstead projection
663/// migrate`. The `mediums/` / `facets/` / `ingests/` trees are **never read**
664/// by this path (a v2 binding carries everything inline); they are served
665/// only by [`load_legacy_pipeline_configs`] for migration.
666pub fn load_pipeline_configs(workspace_root: &Path) -> Result<BindingConfigs, StoreError> {
667    let (bindings, quarantined) = load_bindings(workspace_root)?;
668    Ok(BindingConfigs {
669        bindings,
670        quarantined,
671    })
672}
673
674/// Strict form for WRITE paths (the pipeline-edit layer): a store
675/// carrying ANY quarantined binding refuses with the first entry's
676/// underlying [`StoreError`] — the edit layer never writes over a
677/// store that needs `memstead projection migrate` (or hand repair).
678/// Read/serve paths use the quarantining [`load_pipeline_configs`].
679pub fn load_pipeline_configs_strict(workspace_root: &Path) -> Result<BindingConfigs, StoreError> {
680    let configs = load_pipeline_configs(workspace_root)?;
681    if let Some(q) = configs.quarantined.first() {
682        let path = PathBuf::from(&q.path);
683        return Err(match q.reason_code.as_str() {
684            "UNKNOWN_BINDING_VERSION" => StoreError::UnknownBindingVersion {
685                path,
686                version: q.unknown_version.unwrap_or(-1),
687            },
688            "WORKSPACE_STORE_PARSE" => StoreError::Parse {
689                path,
690                message: q.parse_message.clone().unwrap_or_default(),
691            },
692            _ => StoreError::LegacyProjectionStore { path },
693        });
694    }
695    Ok(configs)
696}
697
698/// Migrate-local: one stored projection file classified by generation, so
699/// the migrate command can route each file to its conversion leg.
700#[derive(Debug, Clone, PartialEq)]
701pub enum ProjectionGeneration {
702    /// Already the live v2 single-record format — nothing to do (idempotence).
703    V2,
704    /// A v1 three-file-store binding — the fold leg converts it. Boxed so
705    /// the enum stays small beside the data-less variants (clippy
706    /// large_enum_variant).
707    V1(Box<crate::binding_migrate::LegacyBindingV1>),
708    /// A version-less gen-2 projection. The ingest-driven gen-2 leg covers
709    /// the ones an ingest schedules; a leftover here is inert (nothing ever
710    /// ran it) and the migrate command surfaces it rather than guessing.
711    VersionLess,
712}
713
714/// Migrate-local: classify every `projections/<mem>/<name>.json` by
715/// generation ([`ProjectionGeneration`]), sorted by `(mem, name)`. Unlike the
716/// live loader this never refuses on version — it exists so `memstead
717/// projection migrate` can see the whole store, whatever its generation. A
718/// malformed file still surfaces [`StoreError::Parse`].
719pub fn load_projection_generations(
720    workspace_root: &Path,
721) -> Result<Vec<(String, String, ProjectionGeneration)>, StoreError> {
722    let raw: Vec<MemPipelineRecord<serde_json::Value>> =
723        load_mem_scoped(workspace_root, PROJECTIONS_DIR)?;
724    let mut out = Vec::with_capacity(raw.len());
725    for record in raw {
726        let generation = match record.config.get("version").and_then(|v| v.as_i64()) {
727            Some(v) if v != 1 && v != 2 => {
728                // A version this engine has never written — refuse rather
729                // than misclassify it as a known generation.
730                return Err(StoreError::UnknownBindingVersion {
731                    path: mem_scoped_path(
732                        workspace_root,
733                        PROJECTIONS_DIR,
734                        &record.mem,
735                        &record.name,
736                    )
737                    .unwrap_or_default(),
738                    version: v,
739                });
740            }
741            Some(2) => ProjectionGeneration::V2,
742            Some(1) => {
743                let v1: crate::binding_migrate::LegacyBindingV1 =
744                    serde_json::from_value(record.config).map_err(|e| StoreError::Parse {
745                        path: mem_scoped_path(
746                            workspace_root,
747                            PROJECTIONS_DIR,
748                            &record.mem,
749                            &record.name,
750                        )
751                        .unwrap_or_default(),
752                        message: e.to_string(),
753                    })?;
754                ProjectionGeneration::V1(Box::new(v1))
755            }
756            _ => ProjectionGeneration::VersionLess,
757        };
758        out.push((record.mem, record.name, generation));
759    }
760    Ok(out)
761}
762
763/// Remove the retired `mediums/` and `facets/` trees after a successful
764/// v1→v2 fold (migrate-local). Tolerates an already-absent tree (idempotent
765/// re-run); any other IO failure surfaces typed.
766pub fn remove_mediums_and_facets_trees(workspace_root: &Path) -> Result<(), StoreError> {
767    for primitive in [MEDIUMS_DIR, FACETS_DIR] {
768        let dir = primitive_dir(workspace_root, primitive);
769        match std::fs::remove_dir_all(&dir) {
770            Ok(()) => {}
771            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
772            Err(e) => {
773                return Err(StoreError::Io {
774                    path: dir,
775                    source: e,
776                });
777            }
778        }
779    }
780    Ok(())
781}
782
783#[cfg(test)]
784mod tests {
785    use super::*;
786    use crate::pipeline::{MediumType, PatternEntry, PatternMode};
787    use tempfile::TempDir;
788
789    fn sample() -> (Medium, Facet, Projection, LegacyIngest) {
790        let medium = Medium {
791            name: "source-tree".to_string(),
792            medium_type: MediumType::Codebase,
793            pointer: "../macos".to_string(),
794            change_detection: None,
795        };
796        let facet = Facet {
797            name: "source-files".to_string(),
798            medium: "source-tree".to_string(),
799            scope: vec![PatternEntry {
800                path: "../macos/**/*.swift".to_string(),
801                mode: PatternMode::Allow,
802            }],
803            engagement: None,
804            preparation: None,
805        };
806        let projection = Projection {
807            intent: Some("Swift macOS app source.".to_string()),
808            source_facets: vec!["source-files".to_string()],
809            reference_mems: vec!["engine".to_string()],
810            destination_mem: "macos".to_string(),
811            rules: None,
812        };
813        let ingest = LegacyIngest {
814            projection: "macos/graph".to_string(),
815            mode: LegacyIngestMode::Discovery,
816            trigger: IngestTrigger::Loop,
817            batch_size: 20,
818            deny_paths: vec!["VISION.md".to_string()],
819            post_actions: None,
820        };
821        (medium, facet, projection, ingest)
822    }
823
824    #[test]
825    fn mutations_refuse_traversal_in_mem_and_name() {
826        // Every mutation path-builds from caller-supplied mem/name; a
827        // separator or traversal segment must refuse with a typed error
828        // and leave nothing on disk outside the store.
829        let tmp = TempDir::new().unwrap();
830        let root = tmp.path();
831        let (medium, _, _, ingest) = sample();
832
833        let evil_values = [
834            "..",
835            ".",
836            "",
837            "../escape",
838            "a/b",
839            "a\\b",
840            "..\\up",
841            "c:evil",
842            "nul\0byte",
843        ];
844        for evil in evil_values {
845            assert!(
846                write_medium(root, evil, "ok", &medium).is_err(),
847                "mem '{}' must refuse",
848                evil.escape_default()
849            );
850            assert!(
851                write_medium(root, "ok", evil, &medium).is_err(),
852                "name '{}' must refuse",
853                evil.escape_default()
854            );
855            assert!(write_ingest(root, evil, &ingest).is_err());
856            assert!(delete_medium(root, evil, "ok").is_err());
857            assert!(delete_ingest(root, evil).is_err());
858            assert!(rename_projection(root, evil, "a", "b").is_err());
859            assert!(rename_projection(root, "ok", evil, "b").is_err());
860            assert!(rename_projection(root, "ok", "a", evil).is_err());
861            assert!(rename_ingest(root, evil, "b").is_err());
862            assert!(rename_ingest(root, "a", evil).is_err());
863        }
864
865        // A traversal write must not have escaped: the only thing under
866        // the temp root may be the (empty) store dir, and the parent of
867        // the temp root gained no `escape.json`.
868        assert!(
869            !root.parent().unwrap().join("escape.json").exists(),
870            "no write may land outside the workspace"
871        );
872
873        // Existing valid names keep working.
874        write_medium(root, "macos", "source-tree", &medium).unwrap();
875        assert!(
876            root.join(".memstead/mediums/macos/source-tree.json")
877                .is_file()
878        );
879    }
880
881    #[test]
882    fn empty_store_loads_empty_configs() {
883        let tmp = TempDir::new().unwrap();
884        let configs = load_legacy_pipeline_configs(tmp.path()).unwrap();
885        assert_eq!(configs, PipelineConfigs::default());
886    }
887
888    #[test]
889    fn write_then_load_round_trips_all_four_primitives() {
890        let tmp = TempDir::new().unwrap();
891        let root = tmp.path();
892        let (medium, facet, projection, ingest) = sample();
893
894        write_medium(root, "macos", "source-tree", &medium).unwrap();
895        write_facet(root, "macos", "source-files", &facet).unwrap();
896        write_projection(root, "macos", "graph", &projection).unwrap();
897        write_ingest(root, "macos-graph", &ingest).unwrap();
898
899        // Files land at the documented `.memstead/` locations.
900        assert!(
901            root.join(".memstead/mediums/macos/source-tree.json")
902                .is_file()
903        );
904        assert!(
905            root.join(".memstead/facets/macos/source-files.json")
906                .is_file()
907        );
908        assert!(
909            root.join(".memstead/projections/macos/graph.json")
910                .is_file()
911        );
912        assert!(root.join(".memstead/ingests/macos-graph.json").is_file());
913
914        let configs = load_legacy_pipeline_configs(root).unwrap();
915        assert_eq!(configs.mediums.len(), 1);
916        assert_eq!(configs.mediums[0].mem, "macos");
917        assert_eq!(configs.mediums[0].name, "source-tree");
918        assert_eq!(configs.mediums[0].config, medium);
919        assert_eq!(configs.facets[0].config, facet);
920        assert_eq!(configs.projections[0].config, projection);
921        assert_eq!(configs.ingests.len(), 1);
922        assert_eq!(configs.ingests[0].name, "macos-graph");
923        assert_eq!(configs.ingests[0].config, ingest);
924    }
925
926    #[test]
927    fn load_enumeration_is_sorted_and_per_mem() {
928        let tmp = TempDir::new().unwrap();
929        let root = tmp.path();
930        let (medium, _, _, _) = sample();
931        write_medium(root, "engine", "z-medium", &medium).unwrap();
932        write_medium(root, "engine", "a-medium", &medium).unwrap();
933        write_medium(root, "macos", "m-medium", &medium).unwrap();
934
935        let configs = load_legacy_pipeline_configs(root).unwrap();
936        let keys: Vec<_> = configs
937            .mediums
938            .iter()
939            .map(|r| (r.mem.as_str(), r.name.as_str()))
940            .collect();
941        assert_eq!(
942            keys,
943            vec![
944                ("engine", "a-medium"),
945                ("engine", "z-medium"),
946                ("macos", "m-medium"),
947            ]
948        );
949    }
950
951    #[test]
952    fn malformed_config_surfaces_typed_parse_error_naming_the_file() {
953        let tmp = TempDir::new().unwrap();
954        let root = tmp.path();
955        let bad = root.join(".memstead/mediums/macos");
956        std::fs::create_dir_all(&bad).unwrap();
957        std::fs::write(bad.join("broken.json"), b"{ not valid json").unwrap();
958
959        let err = load_legacy_pipeline_configs(root).unwrap_err();
960        match err {
961            StoreError::Parse { path, .. } => {
962                assert!(path.ends_with("broken.json"), "got {path:?}");
963            }
964            other => panic!("expected Parse error, got {other:?}"),
965        }
966    }
967
968    #[test]
969    fn delete_removes_the_record_and_load_reflects_it() {
970        let tmp = TempDir::new().unwrap();
971        let root = tmp.path();
972        let (medium, _, _, ingest) = sample();
973        write_medium(root, "macos", "source-tree", &medium).unwrap();
974        write_ingest(root, "macos-graph", &ingest).unwrap();
975
976        delete_medium(root, "macos", "source-tree").unwrap();
977        delete_ingest(root, "macos-graph").unwrap();
978
979        assert!(
980            !root
981                .join(".memstead/mediums/macos/source-tree.json")
982                .exists()
983        );
984        assert!(!root.join(".memstead/ingests/macos-graph.json").exists());
985        let configs = load_legacy_pipeline_configs(root).unwrap();
986        assert!(configs.mediums.is_empty());
987        assert!(configs.ingests.is_empty());
988    }
989
990    #[test]
991    fn delete_of_missing_record_surfaces_io_error() {
992        let tmp = TempDir::new().unwrap();
993        let err = delete_medium(tmp.path(), "macos", "nope").unwrap_err();
994        match err {
995            StoreError::Io { source, .. } => {
996                assert_eq!(source.kind(), std::io::ErrorKind::NotFound);
997            }
998            other => panic!("expected Io error, got {other:?}"),
999        }
1000    }
1001
1002    #[test]
1003    fn rename_moves_the_record_preserving_config() {
1004        let tmp = TempDir::new().unwrap();
1005        let root = tmp.path();
1006        let (_, _, projection, _) = sample();
1007        write_projection(root, "macos", "old-name", &projection).unwrap();
1008
1009        rename_projection(root, "macos", "old-name", "new-name").unwrap();
1010
1011        assert!(
1012            !root
1013                .join(".memstead/projections/macos/old-name.json")
1014                .exists()
1015        );
1016        let configs = load_legacy_pipeline_configs(root).unwrap();
1017        assert_eq!(configs.projections.len(), 1);
1018        assert_eq!(configs.projections[0].name, "new-name");
1019        assert_eq!(configs.projections[0].config, projection);
1020    }
1021
1022    #[test]
1023    fn rename_refuses_to_clobber_an_existing_target() {
1024        let tmp = TempDir::new().unwrap();
1025        let root = tmp.path();
1026        let (_, _, projection, _) = sample();
1027        write_projection(root, "macos", "a", &projection).unwrap();
1028        write_projection(root, "macos", "b", &projection).unwrap();
1029
1030        let err = rename_projection(root, "macos", "a", "b").unwrap_err();
1031        assert!(matches!(err, StoreError::Other(_)), "got {err:?}");
1032        // Both records survive — nothing was lost.
1033        assert!(root.join(".memstead/projections/macos/a.json").exists());
1034        assert!(root.join(".memstead/projections/macos/b.json").exists());
1035    }
1036
1037    #[test]
1038    fn rename_of_missing_source_surfaces_io_error() {
1039        let tmp = TempDir::new().unwrap();
1040        let err = rename_ingest(tmp.path(), "missing", "whatever").unwrap_err();
1041        assert!(matches!(err, StoreError::Io { .. }), "got {err:?}");
1042    }
1043
1044    // ── binding v2 loader (version gate) ─────────────────────────────────
1045
1046    fn sample_binding() -> Binding {
1047        use crate::binding::{BINDING_VERSION, BuildMode, BuildOperation, Operations};
1048        use crate::pipeline::{IngestTrigger, Source};
1049        Binding {
1050            version: BINDING_VERSION,
1051            intent: Some("prose".to_string()),
1052            sources: vec![Source {
1053                name: "source-tree".to_string(),
1054                medium_type: MediumType::Codebase,
1055                pointer: "../public".to_string(),
1056                change_detection: None,
1057                scope: vec![PatternEntry {
1058                    path: "../public/**/*.rs".to_string(),
1059                    mode: PatternMode::Allow,
1060                }],
1061                engagement: None,
1062                preparation: None,
1063            }],
1064            reference_mems: vec![],
1065            destination_mem: "engine".to_string(),
1066            deny_paths: vec![],
1067            coverage_semantics: None,
1068            rules: None,
1069            prune: None,
1070            operations: Operations {
1071                build: Some(BuildOperation {
1072                    mode: BuildMode::Discovery,
1073                    trigger: IngestTrigger::Loop,
1074                    batch_size: 20,
1075                    post_actions: None,
1076                }),
1077                sync: None,
1078                verify: None,
1079            },
1080        }
1081    }
1082
1083    #[test]
1084    fn empty_store_loads_empty_binding_configs() {
1085        let tmp = TempDir::new().unwrap();
1086        let configs = load_pipeline_configs(tmp.path()).unwrap();
1087        assert_eq!(configs, BindingConfigs::default());
1088    }
1089
1090    #[test]
1091    fn binding_loader_round_trips_a_v2_binding() {
1092        let tmp = TempDir::new().unwrap();
1093        let root = tmp.path();
1094        let binding = sample_binding();
1095        write_binding(root, "engine", "graph", &binding).unwrap();
1096
1097        let configs = load_pipeline_configs(root).unwrap();
1098        assert_eq!(configs.bindings.len(), 1);
1099        assert_eq!(configs.bindings[0].mem, "engine");
1100        assert_eq!(configs.bindings[0].name, "graph");
1101        assert_eq!(configs.bindings[0].config, binding);
1102    }
1103
1104    #[test]
1105    fn version_less_projection_refuses_with_migrate_naming_error() {
1106        let tmp = TempDir::new().unwrap();
1107        let root = tmp.path();
1108        // A gen-2 (version-less) projection file.
1109        let projection = Projection {
1110            intent: Some("legacy".to_string()),
1111            source_facets: vec!["f".to_string()],
1112            reference_mems: vec![],
1113            destination_mem: "engine".to_string(),
1114            rules: None,
1115        };
1116        write_projection(root, "engine", "graph", &projection).unwrap();
1117
1118        let err = load_pipeline_configs_strict(root).unwrap_err();
1119        match err {
1120            StoreError::LegacyProjectionStore { path } => {
1121                assert!(path.ends_with("graph.json"), "got {path:?}");
1122                assert!(
1123                    err_message(&StoreError::LegacyProjectionStore { path })
1124                        .contains("memstead projection migrate")
1125                );
1126            }
1127            other => panic!("expected LegacyProjectionStore, got {other:?}"),
1128        }
1129    }
1130
1131    /// REFUSAL (plan criterion 2) — a v1 (three-file-store) binding is a known
1132    /// prior generation: the loader never reads it, surfacing the
1133    /// migrate-naming refusal instead of parsing or reinterpreting.
1134    #[test]
1135    fn v1_binding_refuses_with_migrate_naming_error() {
1136        let tmp = TempDir::new().unwrap();
1137        let root = tmp.path();
1138        let dir = root.join(".memstead/projections/engine");
1139        std::fs::create_dir_all(&dir).unwrap();
1140        std::fs::write(
1141            dir.join("graph.json"),
1142            br#"{"version": 1, "source_facets": ["source-tree"], "destination_mem": "engine", "operations": {"build": {"mode": "discovery", "trigger": "loop", "batch_size": 20}}}"#,
1143        )
1144        .unwrap();
1145
1146        let err = load_pipeline_configs_strict(root).unwrap_err();
1147        match err {
1148            StoreError::LegacyProjectionStore { path } => {
1149                assert!(path.ends_with("graph.json"), "got {path:?}");
1150                assert!(
1151                    err_message(&StoreError::LegacyProjectionStore { path })
1152                        .contains("memstead projection migrate")
1153                );
1154            }
1155            other => panic!("expected LegacyProjectionStore, got {other:?}"),
1156        }
1157    }
1158
1159    #[test]
1160    fn unknown_binding_version_refuses() {
1161        let tmp = TempDir::new().unwrap();
1162        let root = tmp.path();
1163        let dir = root.join(".memstead/projections/engine");
1164        std::fs::create_dir_all(&dir).unwrap();
1165        std::fs::write(
1166            dir.join("graph.json"),
1167            br#"{"version": 99, "destination_mem": "engine", "operations": {"build": {"mode": "discovery", "trigger": "loop", "batch_size": 20}}}"#,
1168        )
1169        .unwrap();
1170
1171        let err = load_pipeline_configs_strict(root).unwrap_err();
1172        assert!(
1173            matches!(err, StoreError::UnknownBindingVersion { version: 99, .. }),
1174            "got {err:?}"
1175        );
1176    }
1177
1178    /// The migrate-local tree removal deletes `mediums/` and `facets/` whole
1179    /// and tolerates their absence (idempotent re-run).
1180    #[test]
1181    fn remove_mediums_and_facets_trees_is_idempotent() {
1182        let tmp = TempDir::new().unwrap();
1183        let root = tmp.path();
1184        let (medium, facet, _, _) = sample();
1185        write_medium(root, "macos", "source-tree", &medium).unwrap();
1186        write_facet(root, "macos", "source-files", &facet).unwrap();
1187
1188        remove_mediums_and_facets_trees(root).unwrap();
1189        assert!(!root.join(".memstead/mediums").exists());
1190        assert!(!root.join(".memstead/facets").exists());
1191
1192        // Second run: nothing to remove, still Ok.
1193        remove_mediums_and_facets_trees(root).unwrap();
1194    }
1195
1196    fn err_message(e: &StoreError) -> String {
1197        e.to_string()
1198    }
1199}