From 420d0c7389fcb98d69aa88bc7f8ece2749f70cef Mon Sep 17 00:00:00 2001 From: Adil Shaikh Date: Mon, 24 Aug 2026 16:02:54 +0000 Subject: [PATCH] feat(curate): wire chi:"curate" end to end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `chi:"curate"` has been in the registry since the protocol was written — "manual compaction request" — but humd dropped it in the thrum.recv.todo arm, no worker handled it, and `graft::prune_jsonl` sat implemented and tested with no production caller. Three pieces of the same feature, none of them joined. This joins them. humd — forward Curate to registered workers, same shape as the Cancel and Cleanup arms directly above it. Workers no-op on an unknown sid, so spraying is safe until sid→worker routing exists. nest — `WorkerBee::curate(sid, cwd) -> CurateReport`, defaulting to a no-op so hives with nothing to trim are unaffected. It takes a cwd rather than a live Cell deliberately: `claude -p` exits after every turn, so a curate almost always arrives with no cell running, and the transcript on disk is the state worth trimming. nest-common — handle "curate" in the worker loop and advertise it in the hello's chis. Adds a small sid→cwd LRU populated from prompts, since a curate names a sid but no cwd and the transcript lives under one. claude-cli — implement curate via the existing prune_jsonl: strip thinking blocks, clip oversized tool results, protect the most recent turns. Two tests cover the new path: curate reaching the right transcript and honouring the protection window, and curate on a sid this bee never raised a cell for, which is routine given humd sprays to every worker. cargo check --workspace clean; claude-cli and nest suites pass. --- hives/claude-cli/src/lib.rs | 30 ++++++++++ hives/claude-cli/tests/curate.rs | 97 ++++++++++++++++++++++++++++++++ hives/common/src/serve.rs | 34 ++++++++++- humd/src/lib.rs | 19 ++++++- nest/src/lib.rs | 25 ++++++++ 5 files changed, 202 insertions(+), 3 deletions(-) create mode 100644 hives/claude-cli/tests/curate.rs diff --git a/hives/claude-cli/src/lib.rs b/hives/claude-cli/src/lib.rs index 90716e4a..b87f6fbe 100644 --- a/hives/claude-cli/src/lib.rs +++ b/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 { + 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 { let cli = spec.cli_path.clone() .or_else(|| std::env::var("CLAUDE_CLI_PATH").ok()) diff --git a/hives/claude-cli/tests/curate.rs b/hives/claude-cli/tests/curate.rs new file mode 100644 index 00000000..f233d5cd --- /dev/null +++ b/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> = 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); +} diff --git a/hives/common/src/serve.rs b/hives/common/src/serve.rs index 7f4939d6..15042280 100644 --- a/hives/common/src/serve.rs +++ b/hives/common/src/serve.rs @@ -134,7 +134,7 @@ async fn dial_and_serve( "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( let cells: Arc>> = 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>> = + 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( 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( } } } + "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 diff --git a/humd/src/lib.rs b/humd/src/lib.rs index 77431edc..2611ec1c 100644 --- a/humd/src/lib.rs +++ b/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 = 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) diff --git a/nest/src/lib.rs b/nest/src/lib.rs index e7f3980f..5149a268 100644 --- a/nest/src/lib.rs +++ b/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; + + /// 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 { + Ok(CurateReport::default()) + } } /// Pollen — what a forager bee carries back alongside the text: