feat(cubesys): per-tenant durable stores on disk (Task 2)

TenantConfig::Disk opens each tenant's ConcurrentStore under its own
sanitized subdir of store_dir (path separators -> '_', no traversal escape),
reusing the durable WAL+checkpoint machinery (incl. 49698af delta-path fix).
Registry takes a config; get_or_provision returns io::Result so a disk
open failure surfaces instead of silently falling back to memory.
Tests: tenant_store_is_isolated_on_disk (separate dirs + survives reopen,
B still isolated) + tenant_id_cannot_traverse_store_dir. ./check green.

cubesys/src/tenant.rs

@@ -2,16 +2,20 @@
 //! per-tenant-isolated CUBELinux-2 store (see the plan at
 //! `.hermes/plans/2026-08-11_041500-concurrent-multitenant-db.md`).
 //!
-//! **Task 1 scope (this file):** the [`TenantId`] type and a
-//! [`TenantRegistry`] skeleton that provisions one [`TenantSession`] per
-//! tenant. No daemon behavior changes yet — tenants are isolated in memory
-//! only; disk-backed per-tenant stores land in Task 2.
+//! **Task 1** landed the [`TenantId`] type and a [`TenantRegistry`] skeleton.
+//! **Task 2** added disk-backed per-tenant stores: each tenant's
+//! [`ConcurrentStore`] is opened under its own sanitized directory under
+//! `store_dir/<tenant>/`, so tenants are isolated on disk and survive a
+//! process restart (reusing the durable WAL + checkpoint machinery, including
+//! the `49698af` delta-path fix). Tasks 3-7 wire the daemon, add the
+//! read/write gate, transactions, and owner/grant enforcement.
 
 use std::collections::HashMap;
+use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::{Arc, RwLock};
 
-use crate::store::ConcurrentStore;
+use crate::store::{ConcurrentStore, DurabilityConfig};
 
 /// Identifies a tenant — the PDF's C-axis environment selector (e.g.
 /// `agent-a`, `hermes`, `cloud`). Tenant ids are cheap to copy-compare and
@@ -59,8 +63,41 @@ impl TenantId {
 }
 
 /// One tenant's view of the store. For Task 1 this is an in-memory
-/// [`ConcurrentStore`] only; Task 2 will make it disk-backed and add the
-/// read/write gate, identity stamping, and auth enforcement.
+/// Configuration for how a tenant's store is materialized.
+///
+/// `Memory` keeps the Task 1 in-memory behavior (handy for tests and the
+/// REPL). `Disk` opens the tenant's [`ConcurrentStore`] at
+/// `store_dir/<tenant>/` so the data is isolated on disk and survives a
+/// restart — the real multi-tenant layout.
+#[derive(Clone, Debug)]
+pub enum TenantConfig {
+    /// In-memory, non-durable (Task 1 default).
+    Memory,
+    /// Disk-backed durable store rooted at `store_dir/`. The tenant subdir is
+    /// derived from the (sanitized) tenant id, so the on-disk path can never
+    /// escape `store_dir` regardless of what id the client presents.
+    Disk {
+        /// Root directory under which each tenant gets its own subdirectory.
+        store_dir: PathBuf,
+        /// Durability tuning forwarded to the store's WAL + checkpoint.
+        durability: DurabilityConfig,
+    },
+}
+
+impl TenantConfig {
+    /// Convenience: disk-backed tenant config at `store_dir` with default
+    /// durability.
+    pub fn disk(store_dir: impl Into<PathBuf>) -> Self {
+        TenantConfig::Disk {
+            store_dir: store_dir.into(),
+            durability: DurabilityConfig::default(),
+        }
+    }
+}
+
+/// One tenant's view of the store. Holds a durable `ConcurrentStore` (memory
+/// or disk) plus the tenant's identity. Task 3 adds the client identity
+/// stamp; Tasks 4-7 add the read/write gate, transactions, and auth.
 ///
 /// Holding the store in an `Arc` lets many connection threads share one
 /// tenant's store cheaply, and lets the registry hand out a cheap clone of
@@ -68,57 +105,94 @@ impl TenantId {
 pub struct TenantSession {
     /// The tenant this session belongs to (useful for diagnostics/telemetry).
     pub id: TenantId,
-    /// The tenant-isolated store. Today always in-memory; Task 2 opens it at
-    /// `store_dir/<tenant>/`.
+    /// The tenant-isolated store. Disk-backed when configured; Task 1 used the
+    /// in-memory variant.
     pub store: Arc<ConcurrentStore>,
 }
 
 impl TenantSession {
-    /// Provision a fresh in-memory tenant session (Task 1 stub).
-    fn memory(id: TenantId) -> Self {
-        TenantSession {
-            id,
-            store: Arc::new(ConcurrentStore::memory()),
+    /// Provision a tenant session per `cfg`. `Memory` yields an in-memory
+    /// store; `Disk` opens the tenant's store at `store_dir/<safe-id>/`,
+    /// where `<safe-id>` is the tenant id with every path-separator replaced
+    /// by `_` so a malicious id can't traverse out of `store_dir`.
+    fn open(id: TenantId, cfg: &TenantConfig) -> std::io::Result<Self> {
+        let store = match cfg {
+            TenantConfig::Memory => ConcurrentStore::memory(),
+            TenantConfig::Disk {
+                store_dir,
+                durability,
+            } => {
+                let safe = id.as_str().replace(['/', '\\', '\0'], "_");
+                let tenant_dir = store_dir.join(&safe);
+                let db = tenant_dir.join("cube-store.json");
+                let wal = tenant_dir.join("cube-store.wal");
+                let rec = tenant_dir.join("cube-store.recovery.ndjson");
+                ConcurrentStore::open(
+                    db.to_str().unwrap(),
+                    wal.to_str().unwrap(),
+                    rec.to_str().unwrap(),
+                    *durability,
+                )?
             }
+        };
+        Ok(TenantSession {
+            id,
+            store: Arc::new(store),
+        })
     }
 }
 
 /// The registry that maps tenant ids to their (isolated) sessions.
 ///
-/// Replacements for the single global `Arc<Mutex<Session>>` in `cube-server`:
+/// Replacement for the single global `Arc<Mutex<Session>>` in `cube-server`:
 /// look up a tenant's `TenantSession` here at connection time. Different
 /// tenants run fully in parallel; within a tenant, the `ConcurrentStore`
 /// already serializes its own writes.
 pub struct TenantRegistry {
     tenants: RwLock<HashMap<TenantId, Arc<TenantSession>>>,
+    /// How each (newly provisioned) tenant's store is materialized. Set at
+    /// registry creation; existing sessions keep the config they were opened
+    /// with.
+    config: TenantConfig,
 }
 
 impl TenantRegistry {
-    /// Create an empty registry.
-    pub fn new() -> Self {
+    /// Create an empty registry with the given tenant materialization config.
+    pub fn with_config(config: TenantConfig) -> Self {
         TenantRegistry {
             tenants: RwLock::new(HashMap::new()),
+            config,
         }
     }
 
-    /// Fetch the session for `id`, provisioning an in-memory one on first use.
+    /// Create an empty registry using in-memory tenant stores (Task 1 default).
+    pub fn new() -> Self {
+        Self::with_config(TenantConfig::Memory)
+    }
+
+    /// Fetch the session for `id`, provisioning one on first use per the
+    /// registry's [`TenantConfig`]. A disk open failure propagates to the
+    /// caller (the daemon should reject the connection rather than silently
+    /// fall back to memory).
     ///
     /// Many readers can hold the read lock simultaneously; only a true miss
     /// (provisioning) takes the write lock, and it drops the lock immediately
     /// after inserting so concurrent provisioning of *different* tenants does
     /// not serialize.
-    pub fn get_or_provision(&self, id: TenantId) -> Arc<TenantSession> {
+    pub fn get_or_provision(&self, id: TenantId) -> std::io::Result<Arc<TenantSession>> {
         // Fast path: already provisioned — take only the read lock.
         if let Some(existing) = self.tenants.read().unwrap().get(&id).cloned() {
-            return existing;
+            return Ok(existing);
         }
         // Miss: take the write lock, but re-check in case another thread
         // provisioned the same tenant while we were waiting.
         let mut guard = self.tenants.write().unwrap();
-        let entry = guard
-            .entry(id.clone())
-            .or_insert_with(|| Arc::new(TenantSession::memory(id)));
-        entry.clone()
+        if let Some(existing) = guard.get(&id).cloned() {
+            return Ok(existing);
+        }
+        let session = Arc::new(TenantSession::open(id.clone(), &self.config)?);
+        guard.insert(id, session.clone());
+        Ok(session)
     }
 }
 
@@ -158,9 +232,15 @@ mod tests {
     #[test]
     fn registry_provisions_and_reuses() {
         let reg = TenantRegistry::new();
-        let a = reg.get_or_provision(TenantId::from_str("agent-a").unwrap());
-        let a2 = reg.get_or_provision(TenantId::from_str("agent-a").unwrap());
-        let b = reg.get_or_provision(TenantId::from_str("agent-b").unwrap());
+        let a = reg
+            .get_or_provision(TenantId::from_str("agent-a").unwrap())
+            .unwrap();
+        let a2 = reg
+            .get_or_provision(TenantId::from_str("agent-a").unwrap())
+            .unwrap();
+        let b = reg
+            .get_or_provision(TenantId::from_str("agent-b").unwrap())
+            .unwrap();
 
         // Same tenant returns the SAME Arc (not a new provisioning).
         assert!(Arc::ptr_eq(&a, &a2), "repeat lookup reuses the session");
@@ -173,9 +253,118 @@ mod tests {
     #[test]
     fn tenants_are_independent_stores() {
         let reg = TenantRegistry::new();
-        let a = reg.get_or_provision(TenantId::from_str("agent-a").unwrap());
-        let b = reg.get_or_provision(TenantId::from_str("agent-b").unwrap());
+        let a = reg
+            .get_or_provision(TenantId::from_str("agent-a").unwrap())
+            .unwrap();
+        let b = reg
+            .get_or_provision(TenantId::from_str("agent-b").unwrap())
+            .unwrap();
         // Each session carries its own store handle.
         assert!(!Arc::ptr_eq(&a.store, &b.store));
     }
+
+    // ---- Task 2: disk-backed per-tenant isolation ----
+
+    /// Two tenants must live in separate on-disk directories, and a write to
+    /// tenant A must be invisible to tenant B after both are reopened from
+    /// disk. This is the core "hard per-namespace partition" guarantee.
+    #[test]
+    fn tenant_store_is_isolated_on_disk() {
+        let root = std::env::temp_dir().join(format!(
+            "cubelinux-tenant-isol-{}-{}",
+            std::process::id(),
+            "a1b2"
+        ));
+        let _ = std::fs::remove_dir_all(&root);
+        std::fs::create_dir_all(&root).unwrap();
+
+        let reg = TenantRegistry::with_config(TenantConfig::disk(&root));
+        let a = reg
+            .get_or_provision(TenantId::from_str("agent-a").unwrap())
+            .unwrap();
+        let b = reg
+            .get_or_provision(TenantId::from_str("agent-b").unwrap())
+            .unwrap();
+
+        // Physically separate dirs.
+        let dir_a = root.join("agent-a");
+        let dir_b = root.join("agent-b");
+        assert!(dir_a.is_dir(), "tenant A has its own dir");
+        assert!(dir_b.is_dir(), "tenant B has its own dir");
+
+        // Write a record to A only.
+        let coord = cubecoords::Czyx::new(1, 1, 1, 1);
+        a.store
+            .put_record(coord, &cubecoords::CubeHeader::new(), b"hello-a");
+        // B must NOT see it (different store).
+        assert!(
+            b.store.get_record(&coord).is_none(),
+            "tenant B must not see tenant A's record"
+        );
+
+        // Force durability, then drop both sessions so the on-disk stores
+        // checkpoint and their background threads stop.
+        a.store.checkpoint();
+        b.store.checkpoint();
+        drop(a);
+        drop(b);
+
+        // Reopen from disk under a fresh registry.
+        let reg2 = TenantRegistry::with_config(TenantConfig::disk(&root));
+        let a2 = reg2
+            .get_or_provision(TenantId::from_str("agent-a").unwrap())
+            .unwrap();
+        let b2 = reg2
+            .get_or_provision(TenantId::from_str("agent-b").unwrap())
+            .unwrap();
+
+        // A's record survived the restart; B still has nothing there.
+        assert_eq!(
+            a2.store.get_record(&coord).map(|(_, v)| v),
+            Some(b"hello-a".to_vec()),
+            "tenant A's record persisted on disk"
+        );
+        assert!(
+            b2.store.get_record(&coord).is_none(),
+            "tenant B still isolated after reopen"
+        );
+
+        // Cleanup.
+        let _ = std::fs::remove_dir_all(&root);
+    }
+
+    /// A tenant id containing path separators must be sanitized so its store
+    /// cannot escape the configured `store_dir`.
+    #[test]
+    fn tenant_id_cannot_traverse_store_dir() {
+        let root = std::env::temp_dir().join(format!(
+            "cubelinux-tenant-traverse-{}-{}",
+            std::process::id(),
+            "c3d4"
+        ));
+        let _ = std::fs::remove_dir_all(&root);
+        std::fs::create_dir_all(&root).unwrap();
+
+        let reg = TenantRegistry::with_config(TenantConfig::disk(&root));
+        let _ = reg
+            .get_or_provision(TenantId::from_str("../escapee").unwrap())
+            .unwrap();
+        let _ = reg
+            .get_or_provision(TenantId::from_str("normal").unwrap())
+            .unwrap();
+
+        // The malicious id lands under root/<sanitized>, NOT outside root.
+        assert!(
+            root.join(".._escapee").is_dir(),
+            "sanitized id stays under store_dir"
+        );
+        // The tenant dir must be a child of (contained by) root — i.e. no
+        // traversal escape actually happened.
+        assert!(
+            root.join(".._escapee").starts_with(&root),
+            "no traversal escape occurred"
+        );
+        // Cleanup.
+        let _ = std::fs::remove_dir_all(&root);
+    }
 }