feat(curate): wire chi:"curate" end to end

hives/claude-cli/src/lib.rs

@@ -111,6 +111,36 @@ impl WorkerBee for ClaudeCliWorker {
     fn ephemeral(&self) -> bool { false }
     fn propensity(&self) -> Propensity { Propensity::StatefulSession }
 
+    /// Prune the sid's transcript in place — drop thinking blocks and
+    /// clip oversized tool results, protecting the most recent turns.
+    ///
+    /// The transcript is the session state here, so this works whether or
+    /// not a cell is live: `claude -p` exits after every turn, and the
+    /// next `--resume` reads whatever is on disk.
+    async fn curate(&self, sid: &ids::HumId, cwd: &str) -> Result<nest::CurateReport> {
+        let derived = sid.to_uuid_v5(ids::NS_CLAUDE_SESSION).to_string();
+        let path = graft::session_path(std::path::Path::new(cwd), &derived);
+        if !path.exists() {
+            trace!(sid = %sid, path = %path.display(), "worker.curate.no-transcript");
+            return Ok(nest::CurateReport::default());
+        }
+
+        let pruned = graft::prune_jsonl(&path)?;
+        trace!(
+            sid = %sid,
+            trimmed = pruned.trimmed,
+            stripped = pruned.stripped,
+            bytes_before = pruned.bytes_before,
+            bytes_after = pruned.bytes_after,
+            "worker.curate.pruned"
+        );
+
+        Ok(nest::CurateReport {
+            bytes_before: pruned.bytes_before as u64,
+            bytes_after: pruned.bytes_after as u64,
+        })
+    }
+
     async fn raise(&self, spec: Egg) -> Result<Cell> {
         let cli = spec.cli_path.clone()
             .or_else(|| std::env::var("CLAUDE_CLI_PATH").ok())

hives/claude-cli/tests/curate.rs

@@ -0,0 +1,97 @@
+//! `WorkerBee::curate` — the compute side of `chi:"curate"`.
+//!
+//! Curation works on the transcript on disk, not on a live cell: `claude -p`
+//! exits after every turn, so a curate almost always arrives with nothing
+//! running. These tests spawn no process.
+//!
+//! `HOME` is repointed at a tempdir because `session_path` resolves
+//! `~/.claude/projects/...`.
+
+use std::fs;
+use std::sync::{Mutex, MutexGuard, OnceLock};
+
+use claude_cli::graft::session_path;
+use claude_cli::ClaudeCliWorker;
+use nest::WorkerBee;
+use tempfile::TempDir;
+
+// HOME is process-global and cargo runs these tests on parallel threads.
+// Every test that repoints it holds this guard for the duration of its
+// filesystem work, matching graft_integration.rs.
+fn home_lock() -> MutexGuard<'static, ()> {
+    static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
+    LOCK.get_or_init(|| Mutex::new(()))
+        .lock()
+        .unwrap_or_else(|e| e.into_inner())
+}
+
+fn sandbox() -> (TempDir, MutexGuard<'static, ()>) {
+    let guard = home_lock();
+    let dir = tempfile::tempdir().expect("tempdir");
+    std::env::set_var("HOME", dir.path());
+    (dir, guard)
+}
+
+fn fixture(name: &str) -> std::path::PathBuf {
+    std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("tests")
+        .join("fixtures")
+        .join(name)
+}
+
+/// Place a fixture transcript where `curate` will look for this sid's.
+fn stage_transcript(home: &TempDir, cwd: &str, sid: &ids::HumId, name: &str) -> std::path::PathBuf {
+    let derived = sid.to_uuid_v5(ids::NS_CLAUDE_SESSION).to_string();
+    let path = session_path(std::path::Path::new(cwd), &derived);
+    fs::create_dir_all(path.parent().expect("parent")).expect("mkdir");
+    fs::copy(fixture(name), &path).expect("copy fixture");
+    let _ = home;
+    path
+}
+
+#[tokio::test]
+async fn curate_prunes_the_sid_transcript_in_place() {
+    let (home, _guard) = sandbox();
+
+    let cwd = "/tmp/proj";
+    let sid = ids::HumId::from_foreign("ses_curate_one");
+    let path = stage_transcript(&home, cwd, &sid, "with_thinking.jsonl");
+    let before = fs::metadata(&path).expect("stat").len();
+
+    let report = ClaudeCliWorker.curate(&sid, cwd).await.expect("curate");
+
+    // Byte counts are of the re-serialized entries, not the raw file, so
+    // they track the transcript's size without matching it exactly.
+    assert!(
+        report.bytes_before > 0,
+        "curate must find and measure the transcript for this sid"
+    );
+
+    // The fixture carries only four user turns and the default protection
+    // window keeps four, so nothing is eligible. Proves curate reached the
+    // right file and honoured the protection invariant rather than
+    // silently missing the path.
+    assert_eq!(report.trimmed(), 0);
+    assert_eq!(
+        fs::metadata(&path).expect("stat").len(),
+        before,
+        "a fully protected transcript must survive intact"
+    );
+}
+
+#[tokio::test]
+async fn curate_is_a_no_op_when_no_transcript_exists() {
+    let (_home, _guard) = sandbox();
+
+    let sid = ids::HumId::from_foreign("ses_never_prompted");
+    let report = ClaudeCliWorker
+        .curate(&sid, "/tmp/nonexistent")
+        .await
+        .expect("curate must not error on a missing transcript");
+
+    // A curate can arrive for a sid this bee has never raised a cell for —
+    // humd sprays to every worker. That is not a failure.
+    assert_eq!(report.bytes_before, 0);
+    assert_eq!(report.bytes_after, 0);
+    assert_eq!(report.trimmed(), 0);
+}

hives/common/src/serve.rs

@@ -134,7 +134,7 @@ async fn dial_and_serve<W: WorkerBee + 'static>(
         "protoVersion": thrum_core::THRUM_VERSION,
         "models": &advert.models,
         "propensity": { "statefulness": propensity_str, "wire": &advert.hive },
-        "chis": ["hello", "prompt", "cancel", "tool-result", "chunk", "finish", "error", "tool-call"],
+        "chis": ["hello", "prompt", "cancel", "curate", "tool-result", "chunk", "finish", "error", "tool-call"],
         "source": advert.source.clone().unwrap_or_default(),
     });
     write_half.lock().await.write_all(format!("{}\n", hello).as_bytes()).await?;
@@ -164,6 +164,11 @@ async fn dial_and_serve<W: WorkerBee + 'static>(
     let cells: Arc<Mutex<LruCache<String, CellBundle>>> =
         Arc::new(Mutex::new(LruCache::new(NonZeroUsize::new(MAX_CELLS).unwrap())));
 
+    // sid → cwd, learned from prompts. A curate names a sid but no cwd,
+    // and the transcript it trims lives under one.
+    let sid_cwd: Arc<Mutex<LruCache<String, String>>> =
+        Arc::new(Mutex::new(LruCache::new(NonZeroUsize::new(MAX_CELLS).unwrap())));
+
     let mut reader = BufReader::new(read_half).lines();
     while let Some(line) = reader.next_line().await? {
         if line.is_empty() { continue; }
@@ -193,6 +198,11 @@ async fn dial_and_serve<W: WorkerBee + 'static>(
                 if !forager_tools.is_empty() || !nestler_tools.is_empty() {
                     bridge.set_catalogue(forager_tools, nestler_tools, &provided);
                 }
+                if !sid.is_empty() {
+                    if let Some(cwd) = tone.get("cwd").and_then(Value::as_str) {
+                        sid_cwd.lock().await.put(sid.clone(), cwd.to_string());
+                    }
+                }
                 let worker = worker.clone();
                 let write_half = write_half.clone();
                 let cells = cells.clone();
@@ -215,6 +225,28 @@ async fn dial_and_serve<W: WorkerBee + 'static>(
                     }
                 }
             }
+            "curate" => {
+                if !sid.is_empty() {
+                    let cwd = sid_cwd.lock().await.get(&sid).cloned();
+                    match cwd {
+                        Some(cwd) => {
+                            let hum_sid = ids::HumId::parse(&sid)
+                                .unwrap_or_else(|_| ids::HumId::from_foreign(&sid));
+                            match worker.curate(&hum_sid, &cwd).await {
+                                Ok(report) => trace!(
+                                    sid = %sid,
+                                    trimmed = report.trimmed(),
+                                    "worker.curate.done"
+                                ),
+                                Err(e) => warn!(sid = %sid, err = %e, "worker.curate.failed"),
+                            }
+                        }
+                        // No prompt has named a cwd for this sid yet, so
+                        // there is no transcript of ours to curate.
+                        None => trace!(sid = %sid, "worker.curate.unknown-sid"),
+                    }
+                }
+            }
             "tool-result" => {
                 let call_id = tone.get("callId").and_then(Value::as_str).map(str::to_string);
                 // First try the worker MCP bridge — humfs_* tools

humd/src/lib.rs

@@ -1619,8 +1619,23 @@ impl ToneSink for HumdSink {
                     Err(e) => warn!(client_id, %author, from, err = %e, "backfill.range.failed"),
                 }
             }
-            Some(Chi::Curate)
-            | Some(Chi::ReleasePermit)
+            Some(Chi::Curate) => {
+                if let Some(_sid) = tone.get("sid").and_then(Value::as_str) {
+                    // Forward the curate to all registered workers — same
+                    // shape as cancel/cleanup above. Workers no-op on an
+                    // unknown sid, so spraying is safe until sid→worker
+                    // routing exists.
+                    let workers: Vec<String> = self.manifests.read()
+                        .iter()
+                        .filter(|(_, m)| m.bee.iter().any(|b| b == "worker"))
+                        .map(|(cid, _)| cid.clone())
+                        .collect();
+                    for wc in workers {
+                        self.thrum.thrum_to(&wc, tone.clone());
+                    }
+                }
+            }
+            Some(Chi::ReleasePermit)
             | Some(Chi::TendrilResult)
             | Some(Chi::PetalCell)
             | Some(Chi::Echo)

nest/src/lib.rs

@@ -119,6 +119,20 @@ pub enum Propensity {
     EphemeralPerCall,
 }
 
+/// What a curation trimmed. Byte counts are of the bee's own stored
+/// transcript, whatever form that takes.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct CurateReport {
+    pub bytes_before: u64,
+    pub bytes_after: u64,
+}
+
+impl CurateReport {
+    pub fn trimmed(&self) -> u64 {
+        self.bytes_before.saturating_sub(self.bytes_after)
+    }
+}
+
 /// A WorkerBee raises cells from eggs — the compute-side trait every
 /// commissioned hive implements.
 #[async_trait]
@@ -128,6 +142,17 @@ pub trait WorkerBee: Send + Sync {
         if self.ephemeral() { Propensity::EphemeralPerCall } else { Propensity::StatefulSession }
     }
     async fn raise(&self, egg: Egg) -> Result<Cell>;
+
+    /// Curate the stored transcript for a sid — trim what can go without
+    /// losing the thread. Answers `chi:"curate"`.
+    ///
+    /// Operates on what the bee has persisted, not on a live cell: a
+    /// harness that exits between turns (claude `-p`) still has a
+    /// transcript to curate, and a curate may well arrive with nothing
+    /// running. Default is a no-op for bees holding nothing to trim.
+    async fn curate(&self, _sid: &HumId, _cwd: &str) -> Result<CurateReport> {
+        Ok(CurateReport::default())
+    }
 }
 
 /// Pollen — what a forager bee carries back alongside the text: