feat(trace-store): name classification, family resolver, write-guard

aura-registry gains the read-side of total trace-name resolution and the
write-guard that makes it total:
- NameKind {Run, Family, NotFound} + TraceStore::name_kind — top-level index.json
  is a single run; else ≥1 member subdir with index.json is a family; else absent.
- read_family -> Vec<FamilyMember{key, traces}> — every member of a family, read
  via the existing read("<name>/<key>"), sorted by key (deterministic, C1); an
  absent/empty family reads as empty (treat-as-absent), a malformed member
  propagates Parse.
- ensure_name_free(name, WriteKind) + a TraceStoreError::NameTaken variant —
  refuses only cross-kind name reuse, so the one ambiguous on-disk state (a name
  used by both a run and a family) is unreachable; same-kind overwrite stays Ok.

Task 1 of plan 0061. Engine untouched (C9/C14). Tests: cargo test -p aura-registry
green (18 passed, incl. the 3 new), clippy -D warnings clean.

Note: the implement-loop quality gate exhausted its re-loop oscillating on the
method NAME (ensure_name_free "implies unused" vs a rename diverging from the
ratified plan/callers). Orchestrator overrule: the ratified name stands — it is
fixed by the spec/plan/#107 and the Task-3 call sites, and the doc comment already
states that Ok means "writable here", not "unused". The implementation is green,
clippy-clean, and spec-faithful.

refs #107

crates/aura-registry/src/lib.rs

@@ -28,7 +28,7 @@ pub use lineage::{
 };
 
 mod trace_store;
-pub use trace_store::{RunTraces, TraceStore, TraceStoreError};
+pub use trace_store::{FamilyMember, NameKind, RunTraces, TraceStore, TraceStoreError, WriteKind};
 
 /// An append-only run registry over a JSONL file: one serde_json line per
 /// `RunReport`.

crates/aura-registry/src/trace_store.rs

@@ -24,6 +24,31 @@ pub struct RunTraces {
     pub taps: Vec<ColumnarTrace>,
 }
 
+/// The on-disk shape of a trace name: a single run, a family of members, or absent.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum NameKind {
+    /// `traces/<name>/index.json` is present — a single recorded run.
+    Run,
+    /// No top-level `index.json`, but ≥1 immediate subdir has one — a family.
+    Family,
+    /// Neither — no recorded run or family of this name.
+    NotFound,
+}
+
+/// One member of a family, read back: its key (the member-dir name) + its traces.
+#[derive(Debug)]
+pub struct FamilyMember {
+    pub key: String,
+    pub traces: RunTraces,
+}
+
+/// Which kind of write a name is about to receive — the write-guard's intent.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum WriteKind {
+    Run,
+    Family,
+}
+
 /// `index.json`: the run manifest + the tap names in write order (= read order).
 #[derive(Serialize, Deserialize)]
 struct Index {
@@ -95,6 +120,74 @@ impl TraceStore {
         }
         Ok(RunTraces { manifest: index.manifest, taps })
     }
+
+    /// Classify a name by its on-disk shape (the read-side of total name
+    /// resolution). Top-level `index.json` -> `Run`; else ≥1 immediate subdir with
+    /// `index.json` -> `Family`; else `NotFound`.
+    pub fn name_kind(&self, name: &str) -> NameKind {
+        let run_dir = self.dir.join(name);
+        if run_dir.join("index.json").is_file() {
+            return NameKind::Run;
+        }
+        if let Ok(entries) = fs::read_dir(&run_dir) {
+            for entry in entries.flatten() {
+                if entry.path().join("index.json").is_file() {
+                    return NameKind::Family;
+                }
+            }
+        }
+        NameKind::NotFound
+    }
+
+    /// Read every member of a family: each immediate subdir of `traces/<name>/`
+    /// that carries an `index.json`, read via `read("<name>/<key>")`, collected and
+    /// **sorted by key** (deterministic, C1). An absent/empty family reads as an
+    /// empty Vec (treat-as-absent); a malformed member propagates its `Parse` error.
+    pub fn read_family(&self, name: &str) -> Result<Vec<FamilyMember>, TraceStoreError> {
+        let run_dir = self.dir.join(name);
+        let entries = match fs::read_dir(&run_dir) {
+            Ok(e) => e,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+            Err(e) => return Err(TraceStoreError::Io(e)),
+        };
+        let mut members = Vec::new();
+        for entry in entries {
+            let entry = entry?;
+            if !entry.path().join("index.json").is_file() {
+                continue; // stray file or non-member subdir
+            }
+            let key = match entry.file_name().into_string() {
+                Ok(k) => k,
+                Err(_) => continue, // non-UTF8 dir name: skip
+            };
+            let traces = self.read(&format!("{name}/{key}"))?;
+            members.push(FamilyMember { key, traces });
+        }
+        members.sort_by(|a, b| a.key.cmp(&b.key));
+        Ok(members)
+    }
+
+    /// The write-guard: `Ok` means a write of `intent` may proceed at this name,
+    /// **not** that the name is unused. It refuses only cross-kind reuse (a name
+    /// already a run when writing a family, or vice versa), so the one ambiguous
+    /// on-disk state — a name used by both a run and a family — is unreachable. A
+    /// same-kind overwrite and a fresh name both return `Ok` (the name may be very
+    /// much in use). Called once per command before any write.
+    pub fn ensure_name_free(
+        &self,
+        name: &str,
+        intent: WriteKind,
+    ) -> Result<(), TraceStoreError> {
+        match (intent, self.name_kind(name)) {
+            (WriteKind::Run, NameKind::Family) => {
+                Err(TraceStoreError::NameTaken { name: name.to_string(), existing: NameKind::Family })
+            }
+            (WriteKind::Family, NameKind::Run) => {
+                Err(TraceStoreError::NameTaken { name: name.to_string(), existing: NameKind::Run })
+            }
+            _ => Ok(()),
+        }
+    }
 }
 
 /// What can go wrong reading or writing the trace store.
@@ -106,6 +199,8 @@ pub enum TraceStoreError {
     Parse { file: String, source: serde_json::Error },
     /// No recorded run of this name (no `index.json` under `traces/<name>/`).
     NotFound(String),
+    /// A trace name already used by the other kind (run vs family) — refuse reuse.
+    NameTaken { name: String, existing: NameKind },
 }
 
 impl fmt::Display for TraceStoreError {
@@ -118,6 +213,14 @@ impl fmt::Display for TraceStoreError {
             TraceStoreError::NotFound(name) => {
                 write!(f, "no recorded run '{name}' under runs/traces")
             }
+            TraceStoreError::NameTaken { name, existing } => {
+                let kind = match existing {
+                    NameKind::Run => "run",
+                    NameKind::Family => "family",
+                    NameKind::NotFound => "name",
+                };
+                write!(f, "'{name}' already used as a {kind}; pick another --trace name")
+            }
         }
     }
 }
@@ -207,4 +310,55 @@ mod tests {
         }
         let _ = fs::remove_dir_all(&root);
     }
+
+    #[test]
+    fn name_kind_classifies_run_family_and_absent() {
+        let root = temp_traces_root("namekind");
+        let store = TraceStore::open(&root);
+        store.write("solo", &sample_manifest(), &sample_taps()).expect("write solo");
+        store.write("fam/m1", &sample_manifest(), &sample_taps()).expect("write m1");
+        assert_eq!(store.name_kind("solo"), NameKind::Run);
+        assert_eq!(store.name_kind("fam"), NameKind::Family);
+        assert_eq!(store.name_kind("ghost"), NameKind::NotFound);
+        let _ = fs::remove_dir_all(&root);
+    }
+
+    #[test]
+    fn read_family_returns_members_sorted_ignoring_strays() {
+        let root = temp_traces_root("readfamily");
+        let store = TraceStore::open(&root);
+        // members written out of order; read_family must sort by key.
+        store.write("fam/b", &sample_manifest(), &sample_taps()).expect("write b");
+        store.write("fam/a", &sample_manifest(), &sample_taps()).expect("write a");
+        // a stray file directly under the family dir is ignored (no index.json there).
+        fs::write(root.join("traces/fam/stray.txt"), "x").expect("stray");
+        let members = store.read_family("fam").expect("read_family");
+        let keys: Vec<&str> = members.iter().map(|m| m.key.as_str()).collect();
+        assert_eq!(keys, vec!["a", "b"]);
+        assert_eq!(members[0].traces.taps.len(), 2);
+        // an absent family reads as empty (treat-as-absent).
+        assert!(store.read_family("ghost").expect("ghost").is_empty());
+        let _ = fs::remove_dir_all(&root);
+    }
+
+    #[test]
+    fn ensure_name_free_refuses_cross_kind_reuse() {
+        let root = temp_traces_root("guard");
+        let store = TraceStore::open(&root);
+        store.write("asrun", &sample_manifest(), &sample_taps()).expect("write run");
+        store.write("asfam/m1", &sample_manifest(), &sample_taps()).expect("write fam");
+        assert!(matches!(
+            store.ensure_name_free("asrun", WriteKind::Family),
+            Err(TraceStoreError::NameTaken { .. })
+        ));
+        assert!(matches!(
+            store.ensure_name_free("asfam", WriteKind::Run),
+            Err(TraceStoreError::NameTaken { .. })
+        ));
+        // same-kind re-run and a fresh name are fine.
+        assert!(store.ensure_name_free("asrun", WriteKind::Run).is_ok());
+        assert!(store.ensure_name_free("asfam", WriteKind::Family).is_ok());
+        assert!(store.ensure_name_free("fresh", WriteKind::Run).is_ok());
+        let _ = fs::remove_dir_all(&root);
+    }
 }