feat(filesystem): add put_batch primitive to RootFilesystem (1/5)

crates/ironclaw_filesystem/CLAUDE.md

@@ -4,10 +4,28 @@
 There is one trait (`RootFilesystem`), one entry type (`Entry`), one mount
 table (`CompositeRootFilesystem`). Every persistence concern in the workspace
 (secrets, leases, processes, memory documents, project files, event logs,
-engine state, settings, …) lives behind a single set of ops: `put` / `get` /
+engine state, settings, …) lives behind a single set of ops: `put` / `put_batch` / `get` /
 `delete` / `list_dir` / `query` / `ensure_index` / `stat` / `begin` /
 `append` / `tail`.
 
+### `put_batch` availability contract
+
+`put_batch(Vec<BatchPut>)` writes several entries in one call and returns one
+`RecordVersion` per put, in input order. Its atomicity depends on the backend:
+an empty batch is a programmer error (`BackendInfrastructure`); `N == 1` is
+**always** available (it routes through `put`, so even CAS-only backends serve
+it); `N > 1` is **all-or-nothing atomic only on `TxnCapability::MultiKey`
+backends** (Postgres today). The default trait impl opens a `begin` transaction
+over the longest common directory prefix of the batch, so a CAS-only backend
+returns the typed `Unsupported{BeginTxn}` for `N > 1` and writes nothing.
+Callers that require atomic batching MUST gate on `Capability::BatchPut` and
+fall back to per-key CAS when it is absent. `CompositeRootFilesystem::put_batch`
+additionally refuses a batch that straddles mounts (`PathOutsideMount`, nothing
+written) since cross-mount atomicity is impossible. PR-1 ships only the trait
+primitive + default impl; PR-2/PR-3/PR-4 add native `put_batch` overrides
+(Postgres single-statement, libSQL transaction, in-memory snapshot) that flip
+the `N > 1` legs to atomic and advertise `Capability::BatchPut`.
+
 This supersedes the earlier "bytes mount; structured records stay typed"
 boundary recorded in
 `docs/reborn/2026-04-25-storage-catalog-and-placement.md`. The override is
@@ -111,3 +129,14 @@ consumers of the legacy methods — new code should call `put`/`get`/
 - Any change to the trait surface needs an accompanying
   `InMemoryBackend` test demonstrating the new op in
   `src/in_memory.rs::tests`.
+- **Adding a `FilesystemOperation` variant requires a WORKSPACE build, not
+  just `-p ironclaw_filesystem`.** The permission gate `operation_allowed`
+  is an exhaustive `match` (no catch-all) duplicated across crates —
+  currently `src/scoped.rs` AND
+  `crates/ironclaw_first_party_extensions/src/coding/paths.rs`. A new
+  variant compiles here but breaks the downstream copy. Grep
+  `rg "FilesystemOperation::Tail" crates src` to find every exhaustive
+  matcher, add the arm to each, then run `cargo build --workspace` before
+  declaring green. (PR-1 of the put_batch work shipped a green
+  `-p ironclaw_filesystem` while the workspace was broken — this note
+  exists so that does not recur.)

crates/ironclaw_filesystem/src/catalog.rs

@@ -5,9 +5,10 @@ use ironclaw_host_api::VirtualPath;
 
 use crate::backend::{EventRecord, StorageTxn};
 use crate::{
-    BackendCapabilities, BackendId, BackendKind, Capability, CasExpectation, ContentKind, DirEntry,
-    Entry, FileStat, FilesystemError, Filter, IndexPolicy, IndexSpec, Page, RecordVersion,
-    RootFilesystem, SeqNo, StorageClass, VersionedEntry, path_prefix_matches,
+    BackendCapabilities, BackendId, BackendKind, BatchPut, Capability, CasExpectation, ContentKind,
+    DirEntry, Entry, FileStat, FilesystemError, FilesystemOperation, Filter, IndexPolicy,
+    IndexSpec, Page, RecordVersion, RootFilesystem, SeqNo, StorageClass, VersionedEntry,
+    path_prefix_matches,
 };
 
 /// Trusted catalog record for one virtual filesystem mount.
@@ -162,6 +163,7 @@ fn validate_mount_capabilities(
         Capability::IndexFts,
         Capability::IndexVector,
         Capability::Events,
+        Capability::BatchPut,
     ];
     let mut shortfalls: Vec<Capability> = NEW_AXES
         .iter()
@@ -265,6 +267,29 @@ impl RootFilesystem for CompositeRootFilesystem {
         self.matching_mount(path)?.backend.begin(path).await
     }
 
+    // A batch may not straddle mounts: every path must resolve to the same
+    // mount as the first leg, else the whole call fails `PathOutsideMount` and
+    // nothing is written. Identity is compared by pointer on the resolved
+    // `CompositeMount`, so a single resolve per path settles routing.
+    async fn put_batch(&self, puts: Vec<BatchPut>) -> Result<Vec<RecordVersion>, FilesystemError> {
+        if puts.is_empty() {
+            return Err(FilesystemError::BackendInfrastructure {
+                operation: FilesystemOperation::PutBatch,
+                reason: "empty put_batch".to_string(),
+            });
+        }
+        let first_mount = self.matching_mount(&puts[0].path)?;
+        for put in &puts[1..] {
+            let mount = self.matching_mount(&put.path)?;
+            if !std::ptr::eq(mount, first_mount) {
+                return Err(FilesystemError::PathOutsideMount {
+                    path: put.path.clone(),
+                });
+            }
+        }
+        first_mount.backend.put_batch(puts).await
+    }
+
     // ── Event plane ──
 
     async fn append(&self, path: &VirtualPath, payload: Vec<u8>) -> Result<SeqNo, FilesystemError> {
@@ -355,3 +380,116 @@ impl RootFilesystem for CompositeRootFilesystem {
             .await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use ironclaw_host_api::VirtualPath;
+
+    use crate::{
+        BackendCapabilities, BackendId, BackendKind, BatchPut, CasExpectation,
+        CompositeRootFilesystem, ContentKind, Entry, FilesystemError, FilesystemOperation,
+        InMemoryBackend, IndexPolicy, MountDescriptor, RootFilesystem, StorageClass,
+    };
+
+    fn descriptor(root: &str) -> MountDescriptor {
+        MountDescriptor {
+            virtual_root: VirtualPath::new(root).unwrap(),
+            backend_id: BackendId::new(format!("mem{}", root.replace('/', "_"))).unwrap(),
+            backend_kind: BackendKind::MemoryDocuments,
+            storage_class: StorageClass::StructuredRecords,
+            content_kind: ContentKind::StructuredRecord,
+            index_policy: IndexPolicy::NotIndexed,
+            capabilities: BackendCapabilities::in_memory_full(),
+        }
+    }
+
+    fn vp(s: &str) -> VirtualPath {
+        VirtualPath::new(s).unwrap()
+    }
+
+    #[tokio::test]
+    async fn put_batch_single_routes_to_owning_mount() {
+        let mut composite = CompositeRootFilesystem::new();
+        let secrets = Arc::new(InMemoryBackend::new());
+        composite
+            .mount(descriptor("/secrets"), secrets.clone())
+            .unwrap();
+
+        let versions = composite
+            .put_batch(vec![BatchPut {
+                path: vp("/secrets/leases/A"),
+                entry: Entry::bytes(vec![9]),
+                cas: CasExpectation::Absent,
+            }])
+            .await
+            .unwrap();
+        assert_eq!(versions.len(), 1);
+        assert_eq!(
+            secrets
+                .get(&vp("/secrets/leases/A"))
+                .await
+                .unwrap()
+                .unwrap()
+                .entry
+                .body,
+            vec![9]
+        );
+    }
+
+    #[tokio::test]
+    async fn put_batch_cross_mount_rejected_writes_nothing() {
+        let mut composite = CompositeRootFilesystem::new();
+        let secrets = Arc::new(InMemoryBackend::new());
+        let memory = Arc::new(InMemoryBackend::new());
+        composite
+            .mount(descriptor("/secrets"), secrets.clone())
+            .unwrap();
+        composite
+            .mount(descriptor("/memory"), memory.clone())
+            .unwrap();
+
+        let err = composite
+            .put_batch(vec![
+                BatchPut {
+                    path: vp("/secrets/leases/A"),
+                    entry: Entry::bytes(vec![1]),
+                    cas: CasExpectation::Absent,
+                },
+                BatchPut {
+                    path: vp("/memory/docs/B"),
+                    entry: Entry::bytes(vec![2]),
+                    cas: CasExpectation::Absent,
+                },
+            ])
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(err, FilesystemError::PathOutsideMount { .. }),
+            "cross-mount put_batch must be rejected, got {err:?}"
+        );
+        // The composite rejects before delegating, so neither backend wrote.
+        assert!(
+            secrets
+                .get(&vp("/secrets/leases/A"))
+                .await
+                .unwrap()
+                .is_none()
+        );
+        assert!(memory.get(&vp("/memory/docs/B")).await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn put_batch_empty_rejected() {
+        let composite = CompositeRootFilesystem::new();
+        let err = composite.put_batch(Vec::new()).await.unwrap_err();
+        assert!(matches!(
+            err,
+            FilesystemError::BackendInfrastructure {
+                operation: FilesystemOperation::PutBatch,
+                ..
+            }
+        ));
+    }
+}

crates/ironclaw_filesystem/src/in_memory.rs

@@ -1183,4 +1183,140 @@ mod tests {
             .await;
         assert!(ok.is_ok(), "top-level VectorNearest must still work");
     }
+
+    #[tokio::test]
+    async fn put_batch_single_put_succeeds() {
+        // N==1 routes through the plain `put`, so it works on every
+        // backend including the CAS-only in-memory reference.
+        let fs = InMemoryBackend::new();
+        let path = vpath("/secrets/leases/batch_single/L1");
+        let versions = fs
+            .put_batch(vec![crate::BatchPut {
+                path: path.clone(),
+                entry: Entry::bytes(vec![1]),
+                cas: CasExpectation::Absent,
+            }])
+            .await
+            .unwrap();
+        assert_eq!(versions.len(), 1);
+        assert_eq!(versions[0].get(), 1);
+        assert_eq!(fs.get(&path).await.unwrap().unwrap().entry.body, vec![1]);
+    }
+
+    #[tokio::test]
+    async fn put_batch_multi_surfaces_unsupported_begin_txn() {
+        // The in-memory backend supports CAS only (`begin` is Unsupported),
+        // so an N>1 put_batch surfaces the typed `Unsupported{BeginTxn}`
+        // from the default trait impl today. PR-4 adds a native multi-key
+        // override that flips this leg to all-or-nothing atomic.
+        let fs = InMemoryBackend::new();
+        let a = vpath("/secrets/leases/batch_multi/A");
+        let b = vpath("/secrets/leases/batch_multi/B");
+        let err = fs
+            .put_batch(vec![
+                crate::BatchPut {
+                    path: a.clone(),
+                    entry: Entry::bytes(vec![1]),
+                    cas: CasExpectation::Absent,
+                },
+                crate::BatchPut {
+                    path: b.clone(),
+                    entry: Entry::bytes(vec![2]),
+                    cas: CasExpectation::Absent,
+                },
+            ])
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(
+                err,
+                FilesystemError::Unsupported {
+                    operation: FilesystemOperation::BeginTxn,
+                    ..
+                }
+            ),
+            "in-memory N>1 put_batch must be Unsupported until PR-4, got {err:?}"
+        );
+        // begin() failed before any write, so nothing landed.
+        assert!(fs.get(&a).await.unwrap().is_none());
+        assert!(fs.get(&b).await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn put_batch_empty_rejected() {
+        let fs = InMemoryBackend::new();
+        let err = fs.put_batch(Vec::new()).await.unwrap_err();
+        assert!(matches!(
+            err,
+            FilesystemError::BackendInfrastructure {
+                operation: FilesystemOperation::PutBatch,
+                ..
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn put_batch_exceeding_cap_rejected_before_any_write() {
+        // A batch larger than the universal MAX_BATCH_PUTS cap is rejected by
+        // the default trait impl before it ever opens a transaction, so
+        // nothing lands. The cap is shared so PR-2/3/4 reuse this one const.
+        let fs = InMemoryBackend::new();
+        let puts: Vec<crate::BatchPut> = (0..=crate::MAX_BATCH_PUTS)
+            .map(|i| crate::BatchPut {
+                path: vpath(&format!("/secrets/leases/cap/L{i}")),
+                entry: Entry::bytes(vec![i as u8]),
+                cas: CasExpectation::Absent,
+            })
+            .collect();
+        assert!(puts.len() > crate::MAX_BATCH_PUTS);
+        let probe = puts[0].path.clone();
+        let err = fs.put_batch(puts).await.unwrap_err();
+        match err {
+            FilesystemError::BackendInfrastructure {
+                operation: FilesystemOperation::PutBatch,
+                reason,
+            } => assert!(
+                reason.contains("MAX_BATCH_PUTS"),
+                "expected cap reason, got {reason}"
+            ),
+            other => panic!("expected BackendInfrastructure cap error, got {other:?}"),
+        }
+        // Cap check short-circuits before begin(), so the first leg is absent.
+        assert!(fs.get(&probe).await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn put_batch_divergent_roots_rejected_by_default_impl() {
+        // When N>1 legs share no leading path component, the default impl
+        // cannot derive a transaction prefix and surfaces a typed
+        // BackendInfrastructure error (nothing written).
+        let fs = InMemoryBackend::new();
+        let err = fs
+            .put_batch(vec![
+                crate::BatchPut {
+                    path: vpath("/memory/a"),
+                    entry: Entry::bytes(vec![1]),
+                    cas: CasExpectation::Absent,
+                },
+                crate::BatchPut {
+                    path: vpath("/turns/b"),
+                    entry: Entry::bytes(vec![2]),
+                    cas: CasExpectation::Absent,
+                },
+            ])
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(
+                err,
+                FilesystemError::BackendInfrastructure {
+                    operation: FilesystemOperation::PutBatch,
+                    ..
+                }
+            ),
+            "divergent-root batch must surface BackendInfrastructure, got {err:?}"
+        );
+        assert!(fs.get(&vpath("/memory/a")).await.unwrap().is_none());
+        assert!(fs.get(&vpath("/turns/b")).await.unwrap().is_none());
+    }
 }