perf(clone): stream cat-file into import via SDK ImportSession — 0.911s -> 0.754s (2.58x -> 2.22x) on codex

ImportSession holds one pooled connection and the dir-path->ino map across
chunk calls; agentfs clone imports directories up front, then overlaps
blob parsing with bounded-channel import chunks. Also de-flakes
overlay_reads_flag_off test (global counter -> per-inode has_pending).

Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>

Author: factory-ain3sh

.agents/specs/2026-06-11-per-phase-1-5x-roadmap-read-ttls-per-request-cost-native-bulk-ingest.md

@@ -32,7 +32,7 @@ until re-verified against codex.
 
 | Phase | noopen off | Default (WS9) | +uring (opt-in) | Target |
 |---|---|---|---|---|
-| clone (plain FUSE) | 9.63x (3.63s) | 9.48x (3.25s) | 8.81x (3.14s) | ≤1.5x miss; `agentfs clone` 2.58x (07-03 remeasure); floor = whole-state double write (pack+worktree 2x43MB into SQLite); pipelining reaches ~2.0x, not 1.5x |
+| clone (plain FUSE) | 9.63x (3.63s) | 9.48x (3.25s) | 8.81x (3.14s) | ≤1.5x miss; `agentfs clone` 2.22x (07-03, streamed ImportSession pipeline: cat-file hidden under import); floor = whole-state double write (pack+worktree 2x43MB into SQLite); 1.5x unreachable in userspace |
 | checkout | 0.49x | **0.42x** ✓ | 0.42x ✓ | hold |
 | status | 1.10x | **0.93x** ✓ | 0.60x ✓ | ≤1.5x **MET** |
 | read_search | 1.87x | **1.41x** ✓ (p25 1.24, p75 1.63) | 1.37x ✓ | ≤1.5x **MET** |

.agents/specs/2026-06-12-enosys-open-eliminate-open-release-round-trips-via-kernel-no_open.notes.md

@@ -50,3 +50,8 @@ User comment: none
 **Type**: decision
 **Context**: (1) agentfs clone remeasured on codex under current defaults: 0.911s / 2.58x (n=5, verified) — noopen+uring do not help the write/SQLite-bound path. Stage budget: clone-no-checkout 293ms + ls-tree 34ms + cat-file 124ms + import 355-382ms + index 6ms + mount ~90ms. Pipelining cat-file into import (-120ms) and import txn tuning land ~0.7s (~2.0x); 1.5x (0.53s) is blocked by the whole-state-in-DB double content write (pack + worktree, 2x43MB into SQLite vs native raw-FS writes). (2) Edit phase decomposed: the benchmark fsyncs each edited file; per-edit floor = fsync drain txn (~154us) + close-time WRITE RT + 2 stat GETATTRs (the same kernel close-time STATX_BLOCKS invalidation). Exact-shape micro: default 2.5-2.8ms per 8 edits — the <=3ms target is already met at micro level; the codex 6-7ms adds larger appends, deeper lookups, and noise. Deferred SETATTR (AGENTFS_DRAIN_ON_SETATTR=0) wins the micro -30% but is codex-parity for the third time (edit 7ms, total noisier) — stays opt-in.
 **Resolution**: Neither miss is legitimately knockable to threshold in userspace: clone's floor is the double write (pipeline work would buy ~0.2s but cannot cross 1.5x, offered to user), edit's residual is the same kernel invalidation plus the fsync txn floor. Scoreboard annotated; per-phase work concludes with 5 of 8 at/under 1.5x and every miss carrying a named, measured floor.
+
+## 2026-07-03T13:35-07:00 — Clone pipeline streamed: 0.911s -> 0.754s (2.58x -> 2.22x)
+**Type**: milestone
+**Context**: SDK gained `ImportSession` (`begin_import` / `import_chunk` / `finish`): one pooled connection plus the directory-path->ino map persist across chunk calls, so imports can be fed incrementally; `import_entries` is now the buffered one-shot wrapper over it. `agentfs clone` now imports all directories in one up-front chunk, then a producer thread parses `git cat-file --batch` output blob-by-blob and sends 4MB/512-entry chunks down a bounded channel while the async consumer imports them — the 124ms cat-file stage now hides entirely under import (stage timings: stream-import 369ms ~= old import alone). Codex n=5: median 0.754s vs native 0.340s = 2.22x (was 0.911s / 2.58x). Correctness: byte-identical `diff -r` vs native clone, clean `git status`, `agentfs integrity` all-ok (cross-chunk parent nlink bumps included), SDK 168 tests x3 parallel + CLI 109 green, noopen-coherence 6/6 both modes. Also fixed a pre-existing test flake exposed by the refactor's timing shift: `overlay_reads_flag_off_falls_back_to_drain_on_write` asserted equality on the process-global batcher enqueue counter, which races under parallel tests; it now asserts the per-inode `has_pending` state immediately after pwrite (a strictly tighter check).
+**Resolution**: ~2.2x is the streaming landing, consistent with the floor analysis: remaining budget is git-clone-no-checkout ~300ms + import ~355ms (SQLite ingest of 43MB at ~120MB/s) + mount ~90ms; 1.5x (0.53s) stays blocked by the whole-state double content write. Clone work concludes here.

cli/src/cmd/clone.rs

@@ -4,11 +4,14 @@
 //! A regular `git clone` through the mount pays ~9-11 FUSE round trips plus
 //! two SQLite transactions per worktree file. This command instead runs
 //! `git clone --no-checkout` through a temporary mount (pack files are a few
-//! large sequential writes), reads the worktree content out of the object
-//! database with `git ls-tree` + `git cat-file --batch`, bulk-imports it via
-//! `AgentFS::import_entries` (large multi-inode transactions), and fabricates
-//! a git index whose cached stat data matches exactly what the filesystem
-//! serves — so `git status` is clean without re-reading any content.
+//! large sequential writes), then streams the worktree content out of the
+//! object database: a producer thread parses `git ls-tree` + `git cat-file
+//! --batch` output while an [`agentfs_sdk::ImportSession`] consumer bulk
+//! imports each chunk (large multi-inode transactions), so blob decoding
+//! overlaps SQLite writes instead of buffering every blob in memory first.
+//! Finally it fabricates a git index whose cached stat data matches exactly
+//! what the filesystem serves — so `git status` is clean without re-reading
+//! any content.
 //!
 //! Invariants: all state lands in the single database file; nothing is
 //! written to the host filesystem. Limitations (v1): submodules are
@@ -136,35 +139,56 @@ async fn clone_into_mount(
 
     let rows = ls_tree(&repo_dir)?;
     stage("ls-tree");
-    let blobs = cat_file_batch(&repo_dir, &rows)?;
-    stage("cat-file-batch");
 
     let dur = SystemTime::now().duration_since(UNIX_EPOCH)?;
     let timestamp = (dur.as_secs() as i64, dur.subsec_nanos() as i64);
     let uid = unsafe { libc::geteuid() };
     let gid = unsafe { libc::getegid() };
 
-    let entries = build_import_entries(&rows, &blobs)?;
-    let bytes: u64 = entries.iter().map(|e| e.data.len() as u64).sum();
-
     use std::os::unix::fs::MetadataExt;
     let repo_meta = std::fs::metadata(&repo_dir).context("failed to stat repository root")?;
     let dest_parent = repo_meta.ino() as i64;
     let dev = repo_meta.dev();
 
-    let imported = agent
-        .import_entries(
+    let mut session = agent
+        .begin_import(
             dest_parent,
-            &entries,
-            &ImportOptions {
+            ImportOptions {
                 uid,
                 gid,
                 timestamp,
             },
         )
         .await
-        .context("bulk import failed")?;
-    stage("import-entries");
+        .context("failed to begin bulk import")?;
+
+    // All directories go in one up-front chunk so streamed file chunks may
+    // arrive in any order relative to each other.
+    session
+        .import_chunk(&dir_entries(&rows)?)
+        .await
+        .context("bulk import failed (directories)")?;
+
+    let (tx, mut rx) = tokio::sync::mpsc::channel::<Vec<ImportEntry>>(4);
+    let producer = spawn_blob_producer(repo_dir.clone(), &rows, tx)?;
+
+    let mut import_err: Option<anyhow::Error> = None;
+    while let Some(chunk) = rx.recv().await {
+        if let Err(error) = session.import_chunk(&chunk).await {
+            import_err = Some(anyhow::Error::from(error));
+            break;
+        }
+    }
+    drop(rx); // unblocks the producer if the import bailed early
+    let produced = producer
+        .join()
+        .map_err(|_| anyhow::anyhow!("blob producer thread panicked"))?;
+    if let Some(error) = import_err {
+        return Err(error.context("bulk import failed"));
+    }
+    let bytes = produced?;
+    let imported = session.finish();
+    stage("stream-import");
 
     let index = build_index_v2(&rows, &imported, timestamp, uid, gid, dev)?;
     std::fs::write(repo_dir.join(".git").join("index"), index)
@@ -268,28 +292,69 @@ fn ls_tree(repo: &Path) -> Result<Vec<TreeRow>> {
     Ok(rows)
 }
 
-/// Fetch every unique blob via one `git cat-file --batch` process. A writer
-/// thread feeds requests so neither side blocks on a full pipe.
-fn cat_file_batch(repo: &Path, rows: &[TreeRow]) -> Result<HashMap<String, Vec<u8>>> {
-    let unique: Vec<String> = {
-        let mut seen = HashSet::new();
-        rows.iter()
-            .filter(|row| seen.insert(row.sha.as_str()))
-            .map(|row| row.sha.clone())
-            .collect()
-    };
+/// Synthesize one import entry per parent directory, first-seen order.
+/// `ls-tree -r` emits paths in index order, so parents always precede
+/// children. Also validates every row's tree entry mode so the streaming
+/// pipeline never starts for an unsupported repository.
+fn dir_entries(rows: &[TreeRow]) -> Result<Vec<ImportEntry>> {
+    let mut entries = Vec::new();
+    let mut known_dirs: HashSet<&str> = HashSet::new();
+
+    for row in rows {
+        match row.mode {
+            MODE_FILE | MODE_EXEC | MODE_SYMLINK => {}
+            // Tolerate historical non-canonical modes git itself normalizes.
+            other => bail!("unsupported tree entry mode {other:o} for {}", row.path),
+        }
+        let mut offset = 0;
+        while let Some(pos) = row.path[offset..].find('/') {
+            let dir = &row.path[..offset + pos];
+            if known_dirs.insert(dir) {
+                entries.push(ImportEntry {
+                    path: dir.to_string(),
+                    mode: S_IFDIR | 0o755,
+                    data: Vec::new(),
+                });
+            }
+            offset += pos + 1;
+        }
+    }
+    Ok(entries)
+}
+
+/// Producer half of the streaming import: fetch every unique blob via one
+/// `git cat-file --batch` process (a writer thread feeds requests so neither
+/// side blocks on a full pipe), fan each blob out to the tree rows that
+/// reference it, and send bounded chunks of import entries down `tx` as they
+/// accumulate. Returns the total content bytes emitted.
+fn spawn_blob_producer(
+    repo: std::path::PathBuf,
+    rows: &[TreeRow],
+    tx: tokio::sync::mpsc::Sender<Vec<ImportEntry>>,
+) -> Result<std::thread::JoinHandle<Result<u64>>> {
+    // sha -> (path, mode) fanout, plus unique shas in first-seen order.
+    let mut unique: Vec<String> = Vec::new();
+    let mut fanout: HashMap<String, Vec<(String, u32)>> = HashMap::new();
+    for row in rows {
+        let refs = fanout.entry(row.sha.clone()).or_insert_with(|| {
+            unique.push(row.sha.clone());
+            Vec::new()
+        });
+        refs.push((row.path.clone(), row.mode));
+    }
 
     let mut child = Command::new("git")
         .arg("-C")
-        .arg(repo)
+        .arg(&repo)
         .args(["cat-file", "--batch"])
         .stdin(Stdio::piped())
         .stdout(Stdio::piped())
         .stderr(Stdio::null())
         .spawn()
         .context("failed to spawn git cat-file --batch")?;
-
     let mut stdin = child.stdin.take().context("missing cat-file stdin")?;
+    let stdout = child.stdout.take().context("missing cat-file stdout")?;
+
     let requests = unique.clone();
     let writer = std::thread::spawn(move || -> std::io::Result<()> {
         for sha in &requests {
@@ -299,15 +364,49 @@ fn cat_file_batch(repo: &Path, rows: &[TreeRow]) -> Result<HashMap<String, Vec<u
         Ok(())
     });
 
-    let mut blobs = HashMap::with_capacity(unique.len());
-    let mut stdout = BufReader::new(child.stdout.take().context("missing cat-file stdout")?);
-    for sha in &unique {
+    let handle = std::thread::spawn(move || -> Result<u64> {
+        let streamed = stream_blobs(&unique, &mut fanout, stdout, &tx);
+        if streamed.is_err() {
+            // Consumer went away or the stream broke; don't leave a git
+            // process wedged on a dead pipe.
+            let _ = child.kill();
+        }
+        let writer_result = writer
+            .join()
+            .map_err(|_| anyhow::anyhow!("cat-file writer thread panicked"));
+        let status = child.wait()?;
+        let bytes = streamed?;
+        writer_result??;
+        if !status.success() {
+            bail!("git cat-file --batch failed with {status}");
+        }
+        Ok(bytes)
+    });
+    Ok(handle)
+}
+
+/// Parse `cat-file --batch` output blob by blob, emitting bounded chunks.
+fn stream_blobs(
+    unique: &[String],
+    fanout: &mut HashMap<String, Vec<(String, u32)>>,
+    stdout: std::process::ChildStdout,
+    tx: &tokio::sync::mpsc::Sender<Vec<ImportEntry>>,
+) -> Result<u64> {
+    const CHUNK_BYTES: usize = 4 * 1024 * 1024;
+    const CHUNK_ENTRIES: usize = 512;
+
+    let mut stdout = BufReader::new(stdout);
+    let mut chunk: Vec<ImportEntry> = Vec::new();
+    let mut chunk_bytes = 0usize;
+    let mut total_bytes = 0u64;
+
+    for sha in unique {
         let mut header = String::new();
         stdout.read_line(&mut header)?;
         let mut fields = header.trim_end().split(' ');
         let echoed = fields.next().unwrap_or_default();
         let kind = fields.next().unwrap_or_default();
-        if kind == "missing" || echoed != sha {
+        if kind == "missing" || echoed != sha.as_str() {
             bail!("git cat-file returned unexpected header for {sha}: {header}");
         }
         let size: usize = fields
@@ -319,59 +418,32 @@ fn cat_file_batch(repo: &Path, rows: &[TreeRow]) -> Result<HashMap<String, Vec<u
         stdout.read_exact(&mut data)?;
         let mut newline = [0u8; 1];
         stdout.read_exact(&mut newline)?;
-        blobs.insert(sha.clone(), data);
-    }
 
-    writer
-        .join()
-        .map_err(|_| anyhow::anyhow!("cat-file writer thread panicked"))??;
-    let status = child.wait()?;
-    if !status.success() {
-        bail!("git cat-file --batch failed with {status}");
-    }
-    Ok(blobs)
-}
-
-/// Expand tree rows into import entries, synthesizing each parent directory
-/// the first time it is seen. `ls-tree -r` emits paths in index order, so
-/// parents always precede children.
-fn build_import_entries(
-    rows: &[TreeRow],
-    blobs: &HashMap<String, Vec<u8>>,
-) -> Result<Vec<ImportEntry>> {
-    let mut entries = Vec::with_capacity(rows.len());
-    let mut known_dirs: HashSet<String> = HashSet::new();
-
-    for row in rows {
-        let mut offset = 0;
-        while let Some(pos) = row.path[offset..].find('/') {
-            let dir = &row.path[..offset + pos];
-            if known_dirs.insert(dir.to_string()) {
-                entries.push(ImportEntry {
-                    path: dir.to_string(),
-                    mode: S_IFDIR | 0o755,
-                    data: Vec::new(),
-                });
+        let refs = fanout
+            .remove(sha.as_str())
+            .with_context(|| format!("no tree rows reference blob {sha}"))?;
+        let last = refs.len() - 1;
+        for (index, (path, mode)) in refs.into_iter().enumerate() {
+            let data = if index == last {
+                std::mem::take(&mut data)
+            } else {
+                data.clone()
+            };
+            total_bytes += data.len() as u64;
+            chunk_bytes += data.len();
+            chunk.push(ImportEntry { path, mode, data });
+            if chunk_bytes >= CHUNK_BYTES || chunk.len() >= CHUNK_ENTRIES {
+                tx.blocking_send(std::mem::take(&mut chunk))
+                    .map_err(|_| anyhow::anyhow!("import consumer stopped"))?;
+                chunk_bytes = 0;
             }
-            offset += pos + 1;
         }
-
-        let data = blobs
-            .get(&row.sha)
-            .with_context(|| format!("missing blob {} for {}", row.sha, row.path))?
-            .clone();
-        let mode = match row.mode {
-            MODE_FILE | MODE_EXEC | MODE_SYMLINK => row.mode,
-            // Tolerate historical non-canonical modes git itself normalizes.
-            other => bail!("unsupported tree entry mode {other:o} for {}", row.path),
-        };
-        entries.push(ImportEntry {
-            path: row.path.clone(),
-            mode,
-            data,
-        });
     }
-    Ok(entries)
+    if !chunk.is_empty() {
+        tx.blocking_send(chunk)
+            .map_err(|_| anyhow::anyhow!("import consumer stopped"))?;
+    }
+    Ok(total_bytes)
 }
 
 /// Serialize a git index (version 2) whose cached stat data matches exactly