From 13c5fe3c41536e829ff2db15b0a852f68f80334f Mon Sep 17 00:00:00 2001 From: Jamie Pine Date: Sun, 9 Aug 2026 19:47:05 -0700 Subject: [PATCH 1/6] fix(sandbox): metadata access for symlinks and path ancestors - allow file-read-metadata on the top-level /etc, /tmp, /var symlinks so resolver config and the CA bundle resolve through their usual paths - allow metadata on ancestors of allowed paths; git clone stats and mkdirs each leading component and a denied stat surfaces as EPERM where the caller expects EEXIST, killing checkout after a full object transfer - add /private/var/select to the read-only set so every sh invocation stops printing 'Operation not permitted' on stderr --- src/sandbox.rs | 85 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 85 insertions(+) diff --git a/src/sandbox.rs b/src/sandbox.rs index 5eb952e2e..f2d00edad 100644 --- a/src/sandbox.rs +++ b/src/sandbox.rs @@ -144,6 +144,7 @@ const MACOS_READ_ONLY_SYSTEM_PATHS: &[&str] = &[ "/Applications", "/private/etc", "/private/var/run", + "/private/var/select", "/private/tmp", "/etc", "/dev", @@ -870,6 +871,50 @@ impl Sandbox { escape_sbpl_path(&tmp) )); + // The top-level /etc, /tmp, and /var symlinks resolve into /private, + // and the subpath rules above only cover the canonical targets. Path + // lookups that traverse a symlink also need read access to the symlink + // itself — without these, resolver config (/etc/resolv.conf) and the + // sh shim's /var/select/sh are unreachable through their usual paths, + // which breaks DNS for every subprocess. + profile.push_str( + "\n; top-level symlinks into /private\n(allow file-read-metadata (literal \"/etc\") (literal \"/tmp\") (literal \"/var\"))\n", + ); + + // Ancestor directories of allowed paths. Tools stat and mkdir each + // leading component of an absolute path (git clone does both when + // creating leading directories), and a denied stat on an ancestor + // surfaces as EPERM where the caller expects EEXIST, aborting the + // operation. Metadata-only access reveals nothing about siblings. + // Emitted before the data_dir deny below so that deny still wins for + // anything under data_dir. + let mut ancestors: Vec = Vec::new(); + let mut collect_ancestors = |path: &Path| { + for ancestor in path.ancestors().skip(1) { + if ancestor == Path::new("/") { + break; + } + if !ancestors.iter().any(|existing| existing == ancestor) { + ancestors.push(ancestor.to_path_buf()); + } + } + }; + collect_ancestors(&workspace); + if self.tools_bin.exists() { + collect_ancestors(&tools_bin); + } + for path in config.all_writable_paths() { + collect_ancestors(&canonicalize_or_self(path)); + } + collect_ancestors(&tmp); + profile.push_str("\n; ancestors of allowed paths (metadata only)\n"); + for ancestor in &ancestors { + profile.push_str(&format!( + "(allow file-read-metadata (literal \"{}\"))\n", + escape_sbpl_path(ancestor) + )); + } + // Protect data_dir even if it falls under the workspace subtree let data_dir = canonicalize_or_self(&self.data_dir); profile.push_str(&format!( @@ -1014,6 +1059,46 @@ mod tests { assert!(config.passthrough_env.is_empty()); } + #[test] + fn test_sbpl_profile_traversal_rules() { + let config = Arc::new(ArcSwap::from_pointee(SandboxConfig { + writable_paths: vec![PathBuf::from("/Users/example/projects/demo")], + ..SandboxConfig::default() + })); + let sandbox = Sandbox::new_for_test( + config.clone(), + PathBuf::from("/Users/example/.spacebot/agents/main/workspace"), + ); + let profile = sandbox.generate_sbpl_profile(&config.load()); + + // Symlink traversal for the /private-backed top-level paths. + assert!(profile.contains( + "(allow file-read-metadata (literal \"/etc\") (literal \"/tmp\") (literal \"/var\"))" + )); + + // Metadata access on each ancestor of the workspace and writable paths. + for ancestor in [ + "/Users", + "/Users/example", + "/Users/example/.spacebot/agents/main", + "/Users/example/projects", + ] { + assert!( + profile.contains(&format!( + "(allow file-read-metadata (literal \"{ancestor}\"))" + )), + "missing ancestor rule for {ancestor}" + ); + } + + // The data dir deny must come after the ancestor allows so it wins. + let deny_pos = profile.find("; data dir blocked").expect("deny section"); + let ancestors_pos = profile + .find("; ancestors of allowed paths") + .expect("ancestors section"); + assert!(ancestors_pos < deny_pos); + } + #[test] fn test_sandbox_mode_serialization() { #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] From d859edf552a2363ec3b1edb2714c702d4336fa15 Mon Sep 17 00:00:00 2001 From: Jamie Pine Date: Sun, 9 Aug 2026 19:47:15 -0700 Subject: [PATCH 2/6] feat(workers): fork channel history by default, feed transcripts to reflection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Workers now fork the channel's conversation history under their own system prompt, the way branches do — the difference between a worker and a branch is the tools it gets, not the context it has. - WorkerHistoryMode collapses to fork (default) | clean. summary was a stub that logged a warning and handed the worker nothing; recent(n) paid for history without any guarantee the load-bearing part was in the window. clean stays for fan-out and mechanical tasks; detached workers are clean by definition. - fork pre-compacts an oversized history before the worker's first LLM call, via a helper shared with the branch pre-flight. - spawn_worker's description now reflects the conversation's live setting, so the model stops writing lossy context summaries out of habit. - the reflection signal carries completed worker ids instead of a bare bool; the persistence prompt lists them and instructs worker_inspect on each, plus failed predecessors — the lesson lives in what the worker tried, not the summary it returned. per-result transcript cap raised to 2000 bytes for reflection passes. - the 'worker process chatter' exclusion is scoped to memory extraction so it no longer argues against reading transcripts during reflection. --- interface/src/api/schema.d.ts | 18 ++-- interface/src/api/types.ts | 4 +- .../components/ConversationSettingsPanel.tsx | 14 ++- prompts/en/memory_persistence.md.j2 | 13 +++ .../en/tools/spawn_worker_description.md.j2 | 2 +- src/agent/branch.rs | 26 +++--- src/agent/channel.rs | 90 ++++++++++++++++--- src/agent/channel_dispatch.rs | 47 +++++----- src/agent/compactor.rs | 34 +++++++ src/api/portal.rs | 7 +- src/conversation/settings.rs | 29 +++--- src/prompts/engine.rs | 19 +++- src/tools.rs | 15 +++- src/tools/spawn_worker.rs | 21 ++++- src/tools/worker_inspect.rs | 16 +++- 15 files changed, 263 insertions(+), 92 deletions(-) diff --git a/interface/src/api/schema.d.ts b/interface/src/api/schema.d.ts index 3b8a62ada..e5bda42e4 100644 --- a/interface/src/api/schema.d.ts +++ b/interface/src/api/schema.d.ts @@ -4623,14 +4623,16 @@ export interface components { transcript?: components["schemas"]["TranscriptStep"][] | null; worker_type: string; }; - /** @description How much conversation history a worker receives. */ - WorkerHistoryMode: "none" | "summary" | { - /** - * Format: int32 - * @description Last N messages from the parent conversation. - */ - recent: number; - } | "full"; + /** + * @description How much conversation history a worker receives. + * + * Workers fork their channel the way branches do: the difference between a + * worker and a branch is the tools it gets, not the context it has. `Clean` + * is the explicit opt-out for fan-out and mechanical tasks where the + * conversation is noise. + * @enum {string} + */ + WorkerHistoryMode: "fork" | "clean"; WorkerListItem: { channel_id?: string | null; channel_name?: string | null; diff --git a/interface/src/api/types.ts b/interface/src/api/types.ts index e41fe27bf..6f17c68c1 100644 --- a/interface/src/api/types.ts +++ b/interface/src/api/types.ts @@ -122,7 +122,7 @@ export type ConversationSettings = { response_mode?: "active" | "observe" | "mention_only"; save_attachments?: boolean; worker_context?: { - history?: "none" | "summary" | "recent" | "full"; + history?: "fork" | "clean"; memory?: "none" | "ambient" | "tools" | "full"; }; }; @@ -132,7 +132,7 @@ export type ConversationDefaultsResponse = { memory: "full" | "ambient" | "off"; delegation: "standard" | "direct"; worker_context: { - history: "none" | "summary" | "recent" | "full"; + history: "fork" | "clean"; memory: "none" | "ambient" | "tools" | "full"; }; available_models: Array<{ diff --git a/interface/src/components/ConversationSettingsPanel.tsx b/interface/src/components/ConversationSettingsPanel.tsx index a33e77856..88c5135b1 100644 --- a/interface/src/components/ConversationSettingsPanel.tsx +++ b/interface/src/components/ConversationSettingsPanel.tsx @@ -33,7 +33,7 @@ const PRESETS: Array<{ settings: { memory: "full", delegation: "standard", - worker_context: {history: "none", memory: "none"}, + worker_context: {history: "fork", memory: "none"}, }, }, { @@ -43,7 +43,7 @@ const PRESETS: Array<{ settings: { memory: "ambient", delegation: "standard", - worker_context: {history: "none", memory: "none"}, + worker_context: {history: "fork", memory: "none"}, }, }, { @@ -54,7 +54,7 @@ const PRESETS: Array<{ settings: { memory: "off", delegation: "direct", - worker_context: {history: "recent", memory: "tools"}, + worker_context: {history: "fork", memory: "tools"}, }, }, { @@ -64,7 +64,7 @@ const PRESETS: Array<{ settings: { memory: "off", delegation: "standard", - worker_context: {history: "none", memory: "none"}, + worker_context: {history: "clean", memory: "none"}, }, }, ]; @@ -109,10 +109,8 @@ const RESPONSE_MODE_DESCRIPTIONS: Record = { }; const WORKER_HISTORY_OPTIONS = [ - {value: "none", label: "None"}, - {value: "summary", label: "Summary"}, - {value: "recent", label: "Recent (20)"}, - {value: "full", label: "Full"}, + {value: "fork", label: "Full context"}, + {value: "clean", label: "Task only"}, ] as const; const WORKER_MEMORY_OPTIONS = [ diff --git a/prompts/en/memory_persistence.md.j2 b/prompts/en/memory_persistence.md.j2 index 6366e5f86..d2fcd18ae 100644 --- a/prompts/en/memory_persistence.md.j2 +++ b/prompts/en/memory_persistence.md.j2 @@ -66,6 +66,10 @@ This is an automatic process triggered periodically during conversation. You are - Ephemeral task chatter: retries, progress updates, temporary tool output, or worker process chatter. + These exclusions govern memory extraction only. When this pass includes + skill reflection, worker transcripts are in scope there — the retries and + dead ends are the raw material a procedure is distilled from. + 8. Verify stale memory before relying on it. If a recalled memory conflicts with the newer conversation context, treat the older item as stale, avoid propagating it as truth, and capture the latest truth via `updates` or `contradicts`. @@ -75,6 +79,15 @@ it as truth, and capture the latest truth via `updates` or `contradicts`. This session involved substantial work, so this pass also decides whether it produced a reusable procedure. Memory answers "who is the user and what is going on"; skills answer "how do we do this class of task here." A correction like "stop posting walls of text in Discord" is not a fact about the user — it's a standing procedure change, and it belongs in the skill governing that task class. First decide whether anything is worth keeping. Ending with no skill writes is the common case and completely acceptable — but treat a session where the user corrected the agent's procedure as a strong write signal. +{% if reflection_worker_ids %} + +These workers completed since the last reflection pass: +{% for worker_id in reflection_worker_ids %} +- `{{ worker_id }}` +{% endfor %} + +Before deciding, pull their transcripts with `worker_inspect` — the lesson usually lives in what a worker tried, not in the summary it returned to the channel. A worker that succeeded after several failed approaches is the strongest signal there is a procedure worth writing down. Also check `worker_inspect` without an id for failed predecessors of the same task: the trials that make a procedure worth keeping are often in the run that didn't succeed just before the one that did. +{% endif %} If something is worth keeping, follow this order strictly: diff --git a/prompts/en/tools/spawn_worker_description.md.j2 b/prompts/en/tools/spawn_worker_description.md.j2 index aa4b3191e..6d7399246 100644 --- a/prompts/en/tools/spawn_worker_description.md.j2 +++ b/prompts/en/tools/spawn_worker_description.md.j2 @@ -1,3 +1,3 @@ -Spawn an independent worker process. By default uses a built-in agent with {tools} tools. The worker only sees the task description you provide — no conversation history.{opencode_note} +Spawn an independent worker process. By default uses a built-in agent with {tools} tools. {history_note}{opencode_note} If OpenCode is enabled and the task is coding-heavy (multi-file edits, debugging, refactors), set `worker_type` to `"opencode"` and include a `directory`. diff --git a/src/agent/branch.rs b/src/agent/branch.rs index 2d106dfde..76cecf1e1 100644 --- a/src/agent/branch.rs +++ b/src/agent/branch.rs @@ -1,6 +1,5 @@ //! Branch: Fork context for thinking and delegation. -use crate::agent::compactor::estimate_history_tokens; use crate::error::Result; use crate::hooks::SpacebotHook; use crate::llm::SpacebotModel; @@ -284,20 +283,19 @@ impl Branch { /// Removes the oldest 50% of messages when usage exceeds 70%. fn maybe_compact_history(&mut self) { let context_window = **self.deps.runtime_config.context_window.load(); - let estimated = estimate_history_tokens(&self.history); - let usage = estimated as f32 / context_window as f32; - - if usage < 0.70 { - return; - } - - tracing::info!( - branch_id = %self.id, - usage = %format!("{:.0}%", usage * 100.0), - history_len = self.history.len(), - "branch pre-compacting history" + let removed = crate::agent::compactor::precompact_forked_history( + &mut self.history, + context_window, + 0.50, ); - self.compact_history(0.50); + if removed > 0 { + tracing::info!( + branch_id = %self.id, + removed, + history_len = self.history.len(), + "branch pre-compacted forked history" + ); + } } /// Aggressive compaction for overflow recovery. Removes 75% of messages. diff --git a/src/agent/channel.rs b/src/agent/channel.rs index d413af8bd..2daab1a81 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -717,8 +717,11 @@ pub struct Channel { last_persistence_at: std::time::Instant, /// Set when a turn or worker crossed the reflection work threshold. /// Consumed by the next persistence branch, which then also reflects - /// on skills. Atomic because turn processing marks it through `&self`. - reflection_signal: std::sync::atomic::AtomicBool, + /// on skills. The worker ids are handed to that branch so it can pull + /// their transcripts via `worker_inspect` — the lesson usually lives in + /// what the worker tried, not in the summary it returned. A mutex (not + /// an atomic) because turn processing marks it through `&self`. + reflection_signal: std::sync::Mutex, /// When the last skill-reflection pass was spawned, for cooldown. last_reflection_at: Option, /// Branch IDs for silent memory persistence branches (results not injected into history). @@ -752,6 +755,24 @@ pub struct Channel { pub resolved_settings: ResolvedConversationSettings, } +/// What accumulated between skill-reflection passes: whether a turn crossed +/// the tool-iteration threshold, and which workers completed successfully. +/// Drained by the persistence branch that performs the reflection. +#[derive(Debug, Default, Clone)] +struct ReflectionSignal { + /// A channel turn crossed `min_tool_iterations` tool calls. + turn_work: bool, + /// Workers that completed successfully since the last reflection pass, + /// in completion order. + worker_ids: Vec, +} + +impl ReflectionSignal { + fn is_set(&self) -> bool { + self.turn_work || !self.worker_ids.is_empty() + } +} + /// RAII guard that records `message_handling_duration_seconds` when dropped, /// ensuring the metric is observed on every exit path (including early returns /// and `?` error propagation). @@ -913,7 +934,7 @@ impl Channel { message_count: 0, last_persistence_at: std::time::Instant::now(), memory_persistence_branches: HashSet::new(), - reflection_signal: std::sync::atomic::AtomicBool::new(false), + reflection_signal: std::sync::Mutex::new(ReflectionSignal::default()), last_reflection_at: None, branch_reply_targets: HashMap::new(), coalesce_buffer: Vec::new(), @@ -3405,7 +3426,7 @@ impl Channel { if *success { let reflection = self.deps.runtime_config.skills_config.load().reflection; if reflection.enabled { - self.mark_reflection_signal("worker_completed"); + self.mark_reflection_worker(*worker_id); // A worker can finish after the last user turn; check // now so reflection doesn't sit pending until the next // inbound message. @@ -3722,19 +3743,47 @@ impl Channel { if self.id.starts_with("cron") { return; } - let was_set = self + let mut signal = self .reflection_signal - .swap(true, std::sync::atomic::Ordering::Relaxed); + .lock() + .expect("reflection signal lock"); + let was_set = signal.is_set(); + signal.turn_work = true; if !was_set { tracing::debug!(channel_id = %self.id, source, "skill reflection signal set"); } } + /// Record a successfully completed worker for the next reflection pass, + /// which pulls its transcript via `worker_inspect`. + fn mark_reflection_worker(&self, worker_id: WorkerId) { + if self.id.starts_with("cron") { + return; + } + let mut signal = self + .reflection_signal + .lock() + .expect("reflection signal lock"); + let was_set = signal.is_set(); + if !signal.worker_ids.contains(&worker_id) { + signal.worker_ids.push(worker_id); + } + if !was_set { + tracing::debug!( + channel_id = %self.id, + %worker_id, + "skill reflection signal set by worker completion" + ); + } + } + /// Whether the next persistence pass should reflect on skills. fn reflection_due(&self) -> bool { if !self .reflection_signal - .load(std::sync::atomic::Ordering::Relaxed) + .lock() + .expect("reflection signal lock") + .is_set() { return false; } @@ -3816,15 +3865,36 @@ impl Channel { self.message_count = 0; self.last_persistence_at = std::time::Instant::now(); - match spawn_memory_persistence_branch(&self.state, &self.deps, reflection_due).await { + // Snapshot the completed-worker ids for the reflection pass; the + // signal itself is only cleared once the branch actually spawns. + let reflection_worker_ids: Vec = if reflection_due { + self.reflection_signal + .lock() + .expect("reflection signal lock") + .worker_ids + .clone() + } else { + Vec::new() + }; + + match spawn_memory_persistence_branch( + &self.state, + &self.deps, + reflection_due, + &reflection_worker_ids, + ) + .await + { Ok(branch_id) => { // Consume the reflection request only once the branch exists; // a failed spawn leaves the signal set so the next check // retries instead of losing the reflection for a cooldown. if reflection_due { self.last_reflection_at = Some(std::time::Instant::now()); - self.reflection_signal - .store(false, std::sync::atomic::Ordering::Relaxed); + *self + .reflection_signal + .lock() + .expect("reflection signal lock") = ReflectionSignal::default(); } self.memory_persistence_branches.insert(branch_id); tracing::info!( diff --git a/src/agent/channel_dispatch.rs b/src/agent/channel_dispatch.rs index d6a3191eb..4fc5ac7dc 100644 --- a/src/agent/channel_dispatch.rs +++ b/src/agent/channel_dispatch.rs @@ -220,6 +220,7 @@ pub(crate) async fn spawn_memory_persistence_branch( state: &ChannelState, deps: &AgentDeps, skill_reflection: bool, + reflection_worker_ids: &[crate::WorkerId], ) -> std::result::Result { let contract_state = Arc::new(MemoryPersistenceContractState::default()); @@ -227,8 +228,12 @@ pub(crate) async fn spawn_memory_persistence_branch( let routing = deps.runtime_config.routing.load(); let model_name = routing.resolve(ProcessType::Branch, None).to_string(); let tool_use_enforcement = deps.runtime_config.tool_use_enforcement.load(); + let reflection_worker_ids: Vec = reflection_worker_ids + .iter() + .map(|id| id.to_string()) + .collect(); let system_prompt = prompt_engine - .render_memory_persistence_prompt(skill_reflection) + .render_memory_persistence_prompt(skill_reflection, &reflection_worker_ids) .and_then(|prompt| { prompt_engine.maybe_append_tool_use_enforcement( prompt, @@ -730,29 +735,29 @@ async fn spawn_worker_inner( } } - // Inject conversation history if needed + // Fork the channel's conversation history under the worker's own system + // prompt — the same fork semantic branches use. An oversized fork is + // compacted here so the worker's first LLM call doesn't start life in + // overflow recovery. let initial_history: Vec = match worker_context.history { - WorkerHistoryMode::None => Vec::new(), - WorkerHistoryMode::Summary => { - // TODO: Generate an LLM-based summary of conversation history. - tracing::warn!( - "WorkerHistoryMode::Summary is not yet implemented, worker will receive no history" + WorkerHistoryMode::Clean => Vec::new(), + WorkerHistoryMode::Fork => { + let mut history = state.history.read().await.clone(); + let context_window = **state.deps.runtime_config.context_window.load(); + let removed = crate::agent::compactor::precompact_forked_history( + &mut history, + context_window, + 0.50, ); - Vec::new() - } - WorkerHistoryMode::Recent(n) => { - let history = state.history.read().await; + if removed > 0 { + tracing::info!( + channel_id = %state.channel_id, + removed, + history_len = history.len(), + "worker fork pre-compacted history" + ); + } history - .iter() - .rev() - .take(n as usize) - .rev() - .cloned() - .collect() - } - WorkerHistoryMode::Full => { - let history = state.history.read().await; - history.clone() } }; diff --git a/src/agent/compactor.rs b/src/agent/compactor.rs index fb44ec86c..9ca9b0e80 100644 --- a/src/agent/compactor.rs +++ b/src/agent/compactor.rs @@ -287,6 +287,40 @@ async fn run_compaction( Ok(remove_count) } +/// Compact a forked history in place when it already crowds the context +/// window: drop the oldest `fraction` of messages and insert a marker so the +/// fork knows material was removed. Used by branch and worker forks before +/// their first LLM call, so a large parent history doesn't start the fork's +/// life in overflow recovery. Returns the number of messages removed. +pub fn precompact_forked_history( + history: &mut Vec, + context_window: usize, + fraction: f32, +) -> usize { + let estimated = estimate_history_tokens(history); + let usage = estimated as f32 / context_window.max(1) as f32; + if usage < 0.70 { + return 0; + } + + let total = history.len(); + if total <= 4 { + return 0; + } + + let remove_count = ((total as f32 * fraction) as usize) + .max(1) + .min(total.saturating_sub(2)); + history.drain(..remove_count); + + let marker = format!( + "[Forked context compacted: {remove_count} older messages removed to stay within context limits. \ + Continue with the information available.]" + ); + history.insert(0, Message::from(marker)); + remove_count +} + /// Estimate token count for a history using chars/4 heuristic. /// /// This is intentionally rough — it's only used for threshold checks, not billing. diff --git a/src/api/portal.rs b/src/api/portal.rs index e2f6ec3c7..aea078024 100644 --- a/src/api/portal.rs +++ b/src/api/portal.rs @@ -521,12 +521,7 @@ pub(super) async fn conversation_defaults( available_models, memory_modes: vec!["full".to_string(), "ambient".to_string(), "off".to_string()], delegation_modes: vec!["standard".to_string(), "direct".to_string()], - worker_history_modes: vec![ - "none".to_string(), - "summary".to_string(), - "recent".to_string(), - "full".to_string(), - ], + worker_history_modes: vec!["fork".to_string(), "clean".to_string()], worker_memory_modes: vec![ "none".to_string(), "ambient".to_string(), diff --git a/src/conversation/settings.rs b/src/conversation/settings.rs index ca15c78df..5bfc22211 100644 --- a/src/conversation/settings.rs +++ b/src/conversation/settings.rs @@ -59,19 +59,21 @@ impl DelegationMode { } /// How much conversation history a worker receives. +/// +/// Workers fork their channel the way branches do: the difference between a +/// worker and a branch is the tools it gets, not the context it has. `Clean` +/// is the explicit opt-out for fan-out and mechanical tasks where the +/// conversation is noise. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] #[serde(rename_all = "snake_case")] pub enum WorkerHistoryMode { - /// No conversation history (current default). - /// Worker sees only the task description. + /// Full clone of the channel's conversation history under the worker's + /// own system prompt — the same fork semantic branches use. #[default] - None, - /// LLM-generated summary of recent conversation context. - Summary, - /// Last N messages from the parent conversation. - Recent(u32), - /// Full conversation history clone (branch-style). - Full, + Fork, + /// Task description only. Detached workers, which have no channel to + /// fork, are always clean. + Clean, } /// How much memory context a worker receives. @@ -433,7 +435,7 @@ mod tests { memory: MemoryMode::Off, delegation: DelegationMode::Direct, worker_context: WorkerContextMode { - history: WorkerHistoryMode::Recent(20), + history: WorkerHistoryMode::Clean, memory: WorkerMemoryMode::Tools, wiki_write: false, }, @@ -450,10 +452,7 @@ mod tests { assert_eq!(resolved.model, Some("conversation-model".to_string())); assert_eq!(resolved.memory, MemoryMode::Off); assert_eq!(resolved.delegation, DelegationMode::Direct); - assert_eq!( - resolved.worker_context.history, - WorkerHistoryMode::Recent(20) - ); + assert_eq!(resolved.worker_context.history, WorkerHistoryMode::Clean); assert_eq!(resolved.worker_context.memory, WorkerMemoryMode::Tools); } @@ -465,7 +464,7 @@ mod tests { assert_eq!(resolved.model, None); assert_eq!(resolved.memory, MemoryMode::Full); assert_eq!(resolved.delegation, DelegationMode::Standard); - assert_eq!(resolved.worker_context.history, WorkerHistoryMode::None); + assert_eq!(resolved.worker_context.history, WorkerHistoryMode::Fork); assert_eq!(resolved.worker_context.memory, WorkerMemoryMode::None); } } diff --git a/src/prompts/engine.rs b/src/prompts/engine.rs index 421e3d336..bd23023ae 100644 --- a/src/prompts/engine.rs +++ b/src/prompts/engine.rs @@ -319,11 +319,16 @@ impl PromptEngine { /// `skill_reflection` adds the reflection section: the pass also decides /// whether the session produced a reusable procedure worth persisting as /// a skill. - pub fn render_memory_persistence_prompt(&self, skill_reflection: bool) -> Result { + pub fn render_memory_persistence_prompt( + &self, + skill_reflection: bool, + reflection_worker_ids: &[String], + ) -> Result { self.render( "memory_persistence", context! { skill_reflection => skill_reflection, + reflection_worker_ids => reflection_worker_ids, }, ) } @@ -903,17 +908,25 @@ mod tests { let engine = PromptEngine::new("en").expect("prompt engine should build"); let plain = engine - .render_memory_persistence_prompt(false) + .render_memory_persistence_prompt(false, &[]) .expect("persistence prompt should render"); assert!(plain.contains("memory persistence process")); assert!(!plain.contains("## Skill Reflection")); let reflecting = engine - .render_memory_persistence_prompt(true) + .render_memory_persistence_prompt(true, &[]) .expect("reflection prompt should render"); assert!(reflecting.contains("## Skill Reflection")); assert!(reflecting.contains("never the incident")); assert!(reflecting.contains("Never persist")); + assert!(!reflecting.contains("completed since the last reflection pass")); + + let worker_ids = vec!["92ae6824-dd29-4f10-bdbe-8e33b4faa35d".to_string()]; + let with_workers = engine + .render_memory_persistence_prompt(true, &worker_ids) + .expect("reflection prompt with workers should render"); + assert!(with_workers.contains("92ae6824-dd29-4f10-bdbe-8e33b4faa35d")); + assert!(with_workers.contains("worker_inspect")); } #[test] diff --git a/src/tools.rs b/src/tools.rs index 28e531d9c..13299080c 100644 --- a/src/tools.rs +++ b/src/tools.rs @@ -893,6 +893,19 @@ pub fn create_branch_tool_server( task_create = task_create.with_api_state(api.clone()); } + // Reflection passes read transcripts for error text and recovery steps, + // which the default per-result display cap clips. + let mut worker_inspect = WorkerInspectTool::new(run_logger, agent_id.to_string()); + if matches!( + &profile, + BranchToolProfile::MemoryPersistence { + skill_reflection: true, + .. + } + ) { + worker_inspect = worker_inspect.with_result_cap(2000); + } + let mut server = ToolServer::new() .tool(memory_save) .tool(MemoryRecallTool::new(memory_search.clone())) @@ -900,7 +913,7 @@ pub fn create_branch_tool_server( .tool(ChannelRecallTool::new(conversation_logger, channel_store)) .tool(SpacebotDocsTool::new()) .tool(EmailSearchTool::new(runtime_config.clone())) - .tool(WorkerInspectTool::new(run_logger, agent_id.to_string())) + .tool(worker_inspect) .tool(task_create) .tool(TaskListTool::new(task_store.clone(), agent_id.to_string())) .tool(TaskUpdateTool::for_branch(task_store, agent_id.clone())) diff --git a/src/tools/spawn_worker.rs b/src/tools/spawn_worker.rs index 37d5304f9..c7dfb1ffe 100644 --- a/src/tools/spawn_worker.rs +++ b/src/tools/spawn_worker.rs @@ -124,15 +124,34 @@ impl Tool for SpawnWorkerTool { "" }; + // The description reflects the conversation's live worker-context + // setting so the model writes task prompts that match what the worker + // will actually see. + let history_mode = self.state.worker_context_settings.read().await.history; + let (history_note, task_description) = match history_mode { + crate::conversation::settings::WorkerHistoryMode::Fork => ( + "The worker forks this conversation's history, so it already knows everything \ + discussed here — describe the task, not the background.", + "Clear, specific description of what the worker should do. The worker shares \ + this conversation's history — don't restate the background.", + ), + crate::conversation::settings::WorkerHistoryMode::Clean => ( + "The worker only sees the task description you provide — no conversation history.", + "Clear, specific description of what the worker should do. Include all context \ + needed since the worker can't see your conversation.", + ), + }; + let base_description = crate::prompts::text::get("tools/spawn_worker"); let description = base_description .replace("{tools}", &tools_list.join(", ")) + .replace("{history_note}", history_note) .replace("{opencode_note}", opencode_note); let mut properties = serde_json::json!({ "task": { "type": "string", - "description": "Clear, specific description of what the worker should do. Include all context needed since the worker can't see your conversation." + "description": task_description }, "interactive": { "type": "boolean", diff --git a/src/tools/worker_inspect.rs b/src/tools/worker_inspect.rs index 5b7f0afe1..a593e378c 100644 --- a/src/tools/worker_inspect.rs +++ b/src/tools/worker_inspect.rs @@ -13,11 +13,15 @@ use rig::tool::Tool; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +/// Per-tool-result display cap when rendering a transcript. +const DEFAULT_RESULT_CAP: usize = 500; + /// Tool for inspecting worker run transcripts. #[derive(Debug, Clone)] pub struct WorkerInspectTool { run_logger: ProcessRunLogger, agent_id: String, + result_cap: usize, } impl WorkerInspectTool { @@ -25,8 +29,16 @@ impl WorkerInspectTool { Self { run_logger, agent_id, + result_cap: DEFAULT_RESULT_CAP, } } + + /// Raise the per-result display cap. Reflection passes read transcripts + /// for the error text and recovery steps, which the default cap can clip. + pub fn with_result_cap(mut self, cap: usize) -> Self { + self.result_cap = cap; + self + } } #[derive(Debug, thiserror::Error)] @@ -142,10 +154,10 @@ impl Tool for WorkerInspectTool { name, text, .. } => { let label = if name.is_empty() { "tool" } else { name }; - let display = if text.len() > 500 { + let display = if text.len() > self.result_cap { format!( "{}\n[truncated, {} bytes total]", - truncate_utf8_ellipsis(text, 500), + truncate_utf8_ellipsis(text, self.result_cap), text.len() ) } else { From a50e66a3c196399f79346e3d196e06be7c8dfb75 Mon Sep 17 00:00:00 2001 From: Jamie Pine Date: Sun, 9 Aug 2026 20:28:14 -0700 Subject: [PATCH 3/6] fix(workers): address review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - collect failed worker completions in the reflection signal too — their transcripts carry the trials — while keeping success-only as the trigger policy. Entries are handed to the pass annotated succeeded/failed. - precompact_forked_history re-estimates after each drain instead of trusting one fractional cut; a few large messages could previously leave the fork over budget. Stops at a 4-message floor and leaves the rest to overflow recovery. Covered by tests. --- prompts/en/memory_persistence.md.j2 | 2 +- src/agent/channel.rs | 58 ++++++++++-------- src/agent/channel_dispatch.rs | 9 ++- src/agent/compactor.rs | 95 +++++++++++++++++++++++------ 4 files changed, 117 insertions(+), 47 deletions(-) diff --git a/prompts/en/memory_persistence.md.j2 b/prompts/en/memory_persistence.md.j2 index d2fcd18ae..17961dd28 100644 --- a/prompts/en/memory_persistence.md.j2 +++ b/prompts/en/memory_persistence.md.j2 @@ -86,7 +86,7 @@ These workers completed since the last reflection pass: - `{{ worker_id }}` {% endfor %} -Before deciding, pull their transcripts with `worker_inspect` — the lesson usually lives in what a worker tried, not in the summary it returned to the channel. A worker that succeeded after several failed approaches is the strongest signal there is a procedure worth writing down. Also check `worker_inspect` without an id for failed predecessors of the same task: the trials that make a procedure worth keeping are often in the run that didn't succeed just before the one that did. +Before deciding, pull their transcripts with `worker_inspect` — the lesson usually lives in what a worker tried, not in the summary it returned to the channel. A worker that succeeded after failed attempts — its own retries or a failed run listed above — is the strongest signal there is a procedure worth writing down. If the trail extends past this list, `worker_inspect` without an id shows older runs. {% endif %} If something is worth keeping, follow this order strictly: diff --git a/src/agent/channel.rs b/src/agent/channel.rs index 2daab1a81..8e3eaa7a0 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -756,20 +756,24 @@ pub struct Channel { } /// What accumulated between skill-reflection passes: whether a turn crossed -/// the tool-iteration threshold, and which workers completed successfully. -/// Drained by the persistence branch that performs the reflection. +/// the tool-iteration threshold, and which workers completed since the last +/// pass. Drained by the persistence branch that performs the reflection. +/// +/// Failed completions are collected too — their transcripts are where the +/// trials live — but only successful ones make the signal fire: an +/// unresolved failure alone has nothing to teach. #[derive(Debug, Default, Clone)] struct ReflectionSignal { /// A channel turn crossed `min_tool_iterations` tool calls. turn_work: bool, - /// Workers that completed successfully since the last reflection pass, - /// in completion order. - worker_ids: Vec, + /// Workers that completed since the last reflection pass, in completion + /// order, with whether each succeeded. + workers: Vec<(WorkerId, bool)>, } impl ReflectionSignal { fn is_set(&self) -> bool { - self.turn_work || !self.worker_ids.is_empty() + self.turn_work || self.workers.iter().any(|(_, success)| *success) } } @@ -3421,16 +3425,20 @@ impl Channel { run_logger.log_worker_completed(*worker_id, result, *success); - // A worker finishing real work successfully is a reflection - // signal: the session likely produced a reusable lesson. - if *success { + // Every completion is recorded for the next reflection pass — + // failed transcripts carry the trials — but only success + // fires the signal: a worker finishing real work successfully + // means the session likely produced a reusable lesson. + { let reflection = self.deps.runtime_config.skills_config.load().reflection; if reflection.enabled { - self.mark_reflection_worker(*worker_id); - // A worker can finish after the last user turn; check - // now so reflection doesn't sit pending until the next - // inbound message. - self.check_memory_persistence().await; + self.mark_reflection_worker(*worker_id, *success); + if *success { + // A worker can finish after the last user turn; + // check now so reflection doesn't sit pending + // until the next inbound message. + self.check_memory_persistence().await; + } } } @@ -3754,9 +3762,11 @@ impl Channel { } } - /// Record a successfully completed worker for the next reflection pass, - /// which pulls its transcript via `worker_inspect`. - fn mark_reflection_worker(&self, worker_id: WorkerId) { + /// Record a completed worker for the next reflection pass, which pulls + /// its transcript via `worker_inspect`. Failed workers are recorded too + /// — their transcripts carry the trials — but don't fire the signal by + /// themselves. + fn mark_reflection_worker(&self, worker_id: WorkerId, success: bool) { if self.id.starts_with("cron") { return; } @@ -3765,10 +3775,10 @@ impl Channel { .lock() .expect("reflection signal lock"); let was_set = signal.is_set(); - if !signal.worker_ids.contains(&worker_id) { - signal.worker_ids.push(worker_id); + if !signal.workers.iter().any(|(id, _)| *id == worker_id) { + signal.workers.push((worker_id, success)); } - if !was_set { + if !was_set && signal.is_set() { tracing::debug!( channel_id = %self.id, %worker_id, @@ -3865,13 +3875,13 @@ impl Channel { self.message_count = 0; self.last_persistence_at = std::time::Instant::now(); - // Snapshot the completed-worker ids for the reflection pass; the + // Snapshot the completed workers for the reflection pass; the // signal itself is only cleared once the branch actually spawns. - let reflection_worker_ids: Vec = if reflection_due { + let reflection_workers: Vec<(WorkerId, bool)> = if reflection_due { self.reflection_signal .lock() .expect("reflection signal lock") - .worker_ids + .workers .clone() } else { Vec::new() @@ -3881,7 +3891,7 @@ impl Channel { &self.state, &self.deps, reflection_due, - &reflection_worker_ids, + &reflection_workers, ) .await { diff --git a/src/agent/channel_dispatch.rs b/src/agent/channel_dispatch.rs index 4fc5ac7dc..259af01c0 100644 --- a/src/agent/channel_dispatch.rs +++ b/src/agent/channel_dispatch.rs @@ -220,7 +220,7 @@ pub(crate) async fn spawn_memory_persistence_branch( state: &ChannelState, deps: &AgentDeps, skill_reflection: bool, - reflection_worker_ids: &[crate::WorkerId], + reflection_workers: &[(crate::WorkerId, bool)], ) -> std::result::Result { let contract_state = Arc::new(MemoryPersistenceContractState::default()); @@ -228,9 +228,12 @@ pub(crate) async fn spawn_memory_persistence_branch( let routing = deps.runtime_config.routing.load(); let model_name = routing.resolve(ProcessType::Branch, None).to_string(); let tool_use_enforcement = deps.runtime_config.tool_use_enforcement.load(); - let reflection_worker_ids: Vec = reflection_worker_ids + let reflection_worker_ids: Vec = reflection_workers .iter() - .map(|id| id.to_string()) + .map(|(id, success)| { + let status = if *success { "succeeded" } else { "failed" }; + format!("{id} — {status}") + }) .collect(); let system_prompt = prompt_engine .render_memory_persistence_prompt(skill_reflection, &reflection_worker_ids) diff --git a/src/agent/compactor.rs b/src/agent/compactor.rs index 9ca9b0e80..8e3378512 100644 --- a/src/agent/compactor.rs +++ b/src/agent/compactor.rs @@ -297,28 +297,39 @@ pub fn precompact_forked_history( context_window: usize, fraction: f32, ) -> usize { - let estimated = estimate_history_tokens(history); - let usage = estimated as f32 / context_window.max(1) as f32; - if usage < 0.70 { - return 0; - } + let mut removed_total = 0usize; + + // Re-estimate after each drain: one fractional cut isn't guaranteed to + // land under budget when a few large messages dominate the history. The + // floor of 4 retained messages bounds the loop; a history that still + // exceeds the budget at the floor is left for overflow recovery. + loop { + let estimated = estimate_history_tokens(history); + let usage = estimated as f32 / context_window.max(1) as f32; + if usage < 0.70 { + break; + } - let total = history.len(); - if total <= 4 { - return 0; - } + let total = history.len(); + if total <= 4 { + break; + } - let remove_count = ((total as f32 * fraction) as usize) - .max(1) - .min(total.saturating_sub(2)); - history.drain(..remove_count); + let remove_count = ((total as f32 * fraction) as usize) + .max(1) + .min(total.saturating_sub(2)); + history.drain(..remove_count); + removed_total += remove_count; + } - let marker = format!( - "[Forked context compacted: {remove_count} older messages removed to stay within context limits. \ - Continue with the information available.]" - ); - history.insert(0, Message::from(marker)); - remove_count + if removed_total > 0 { + let marker = format!( + "[Forked context compacted: {removed_total} older messages removed to stay within context limits. \ + Continue with the information available.]" + ); + history.insert(0, Message::from(marker)); + } + removed_total } /// Estimate token count for a history using chars/4 heuristic. @@ -474,3 +485,49 @@ pub enum CompactionAction { /// Emergency truncation (no LLM, drop oldest 50%). EmergencyTruncate, } + +#[cfg(test)] +mod tests { + use super::*; + + fn text_message(size: usize) -> Message { + Message::from("x".repeat(size)) + } + + #[test] + fn precompact_noop_under_budget() { + let mut history: Vec = (0..10).map(|_| text_message(100)).collect(); + let removed = precompact_forked_history(&mut history, 200_000, 0.50); + assert_eq!(removed, 0); + assert_eq!(history.len(), 10); + } + + #[test] + fn precompact_drains_until_under_budget() { + // 40 messages of 4000 chars ≈ 40k tokens against a 20k window: a + // single 50% cut leaves ~20k (still ≥70%), so the loop must run + // more than once. + let mut history: Vec = (0..40).map(|_| text_message(4000)).collect(); + let removed = precompact_forked_history(&mut history, 20_000, 0.50); + assert!(removed > 20, "one fractional cut is not enough: {removed}"); + let estimated = estimate_history_tokens(&history); + assert!((estimated as f32) < 20_000.0 * 0.70); + // Marker inserted once, at the front. + let front = match &history[0] { + Message::User { content } => format!("{content:?}"), + other => format!("{other:?}"), + }; + assert!(front.contains("Forked context compacted")); + assert!(!format!("{:?}", &history[1]).contains("Forked context compacted")); + } + + #[test] + fn precompact_stops_at_retention_floor() { + // A handful of giant messages that can never fit the budget: the + // loop must stop at the floor instead of spinning or emptying. + let mut history: Vec = (0..6).map(|_| text_message(100_000)).collect(); + let removed = precompact_forked_history(&mut history, 10_000, 0.50); + assert!(history.len() >= 4); + assert!(removed > 0); + } +} From 5c96879a754f90c5f53de972b235ea5f9941f6cc Mon Sep 17 00:00:00 2001 From: Jamie Pine Date: Sun, 9 Aug 2026 20:40:26 -0700 Subject: [PATCH 4/6] fix(tests): drop redundant borrow flagged by clippy --- src/agent/compactor.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/agent/compactor.rs b/src/agent/compactor.rs index 8e3378512..82510e812 100644 --- a/src/agent/compactor.rs +++ b/src/agent/compactor.rs @@ -518,7 +518,7 @@ mod tests { other => format!("{other:?}"), }; assert!(front.contains("Forked context compacted")); - assert!(!format!("{:?}", &history[1]).contains("Forked context compacted")); + assert!(!format!("{:?}", history[1]).contains("Forked context compacted")); } #[test] From 5d353b9e16d063ddbbcecd0336cabfb95e62f105 Mon Sep 17 00:00:00 2001 From: Jamie Pine Date: Sun, 9 Aug 2026 22:09:55 -0700 Subject: [PATCH 5/6] fix(compactor): budget forked history against reserved prompt capacity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pre-compaction stopped at a raw 70% usage check against the whole context window, so a fork that fit still had no room for its own system prompt and response. Budget history against the window minus a 30% reserve, re-estimate after every cut, and count the compaction marker itself against the budget. The retention floor is now a named constant and applies to the drain clamp too (it previously allowed cutting to 2 while the loop guarded at 4). A fork whose retained tail alone busts the budget — one oversized tool result can — keeps that tail and logs a warning rather than silently returning over budget. Reflection worker recording moves onto ReflectionSignal so the lifecycle is testable: failed workers are collected for their transcripts but don't fire the signal alone, and repeat completions for a worker are ignored. --- src/agent/branch.rs | 4 +- src/agent/channel.rs | 73 ++++++++++++++++++-- src/agent/compactor.rs | 151 ++++++++++++++++++++++++++++++++++------- 3 files changed, 199 insertions(+), 29 deletions(-) diff --git a/src/agent/branch.rs b/src/agent/branch.rs index 76cecf1e1..d913fa339 100644 --- a/src/agent/branch.rs +++ b/src/agent/branch.rs @@ -279,8 +279,8 @@ impl Branch { Ok(conclusion) } - /// Compact history if approaching context window limit. - /// Removes the oldest 50% of messages when usage exceeds 70%. + /// Compact history down to the fork's token budget, dropping the oldest + /// half of the messages at a time until it fits. fn maybe_compact_history(&mut self) { let context_window = **self.deps.runtime_config.context_window.load(); let removed = crate::agent::compactor::precompact_forked_history( diff --git a/src/agent/channel.rs b/src/agent/channel.rs index f2c29f361..b86bde752 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -892,6 +892,16 @@ impl ReflectionSignal { fn is_set(&self) -> bool { self.turn_work || self.workers.iter().any(|(_, success)| *success) } + + /// Record a completed worker for the next reflection pass. Repeat + /// completions for the same worker are ignored so a retriggered event + /// can't queue the same transcript twice. + fn record_worker(&mut self, worker_id: WorkerId, success: bool) { + if self.workers.iter().any(|(id, _)| *id == worker_id) { + return; + } + self.workers.push((worker_id, success)); + } } /// RAII guard that records `message_handling_duration_seconds` when dropped, @@ -3998,9 +4008,7 @@ impl Channel { .lock() .expect("reflection signal lock"); let was_set = signal.is_set(); - if !signal.workers.iter().any(|(id, _)| *id == worker_id) { - signal.workers.push((worker_id, success)); - } + signal.record_worker(worker_id, success); if !was_set && signal.is_set() { tracing::debug!( channel_id = %self.id, @@ -4368,7 +4376,7 @@ fn is_dm_conversation_id(conv_id: &str) -> bool { #[cfg(test)] mod tests { use super::{ - ObserveModeFallbackState, branch_working_memory_event_summary, + ObserveModeFallbackState, ReflectionSignal, branch_working_memory_event_summary, classify_conversational_event_summary, compute_listen_mode_invocation, decision_user_id, extract_decision_summary_from_reply, format_conversational_event_summary, is_dm_conversation_id, recv_channel_event, should_process_event_for_channel, @@ -4403,6 +4411,63 @@ mod tests { } } + #[test] + fn reflection_signal_records_failed_workers_without_firing() { + let mut signal = ReflectionSignal::default(); + let failed = uuid::Uuid::new_v4(); + signal.record_worker(failed, false); + + assert_eq!(signal.workers, vec![(failed, false)]); + assert!( + !signal.is_set(), + "a failure on its own has no lesson to reflect on" + ); + } + + #[test] + fn reflection_signal_carries_failed_predecessors_of_a_success() { + let mut signal = ReflectionSignal::default(); + let first_failure = uuid::Uuid::new_v4(); + let second_failure = uuid::Uuid::new_v4(); + let success = uuid::Uuid::new_v4(); + signal.record_worker(first_failure, false); + signal.record_worker(second_failure, false); + signal.record_worker(success, true); + + assert!(signal.is_set()); + assert_eq!( + signal.workers, + vec![ + (first_failure, false), + (second_failure, false), + (success, true), + ], + "reflection needs the failed attempts that preceded the success" + ); + } + + #[test] + fn reflection_signal_fires_on_turn_work_alone() { + let mut signal = ReflectionSignal::default(); + assert!(!signal.is_set()); + signal.turn_work = true; + assert!(signal.is_set()); + } + + #[test] + fn reflection_signal_ignores_repeat_completions() { + let mut signal = ReflectionSignal::default(); + let worker = uuid::Uuid::new_v4(); + signal.record_worker(worker, false); + signal.record_worker(worker, true); + + assert_eq!( + signal.workers, + vec![(worker, false)], + "the first completion is the terminal one" + ); + } + #[tokio::test] async fn channel_event_loop_continues_after_lagged_broadcast() { let (event_tx, mut event_rx) = tokio::sync::broadcast::channel::(2); diff --git a/src/agent/compactor.rs b/src/agent/compactor.rs index 82510e812..402804b4b 100644 --- a/src/agent/compactor.rs +++ b/src/agent/compactor.rs @@ -287,47 +287,86 @@ async fn run_compaction( Ok(remove_count) } -/// Compact a forked history in place when it already crowds the context -/// window: drop the oldest `fraction` of messages and insert a marker so the +/// Share of the context window pre-compaction holds back for the fork's system +/// prompt and the model's response. Forked history is budgeted against what +/// remains, so a fork that fits still has room to make its first call. +const FORK_CONTEXT_RESERVE: f32 = 0.30; + +/// Messages pre-compaction never drops. A fork needs its most recent exchange +/// to be able to act at all, so the budget yields to it rather than the +/// reverse. +const FORK_MIN_RETAINED_MESSAGES: usize = 4; + +/// Tokens available to a forked history, once the fork's own prompt and +/// response capacity is reserved. +fn forked_history_budget(context_window: usize) -> usize { + let reserve = (context_window as f32 * FORK_CONTEXT_RESERVE) as usize; + context_window.saturating_sub(reserve) +} + +fn forked_compaction_marker(removed_total: usize) -> String { + format!( + "[Forked context compacted: {removed_total} older messages removed to stay within context limits. \ + Continue with the information available.]" + ) +} + +/// Upper bound on what the marker itself costs, counted against the budget so +/// the notice can't push a fork back over the line it was just trimmed to. +fn forked_compaction_marker_tokens(removed_total: usize) -> usize { + if removed_total == 0 { + return 0; + } + forked_compaction_marker(removed_total).len().div_ceil(4) +} + +/// Compact a forked history in place until it fits the fork's token budget: +/// drop the oldest `fraction` of messages at a time and insert a marker so the /// fork knows material was removed. Used by branch and worker forks before /// their first LLM call, so a large parent history doesn't start the fork's /// life in overflow recovery. Returns the number of messages removed. +/// +/// The most recent [`FORK_MIN_RETAINED_MESSAGES`] messages are preserved even +/// when they alone exceed the budget — a single oversized tool result can +/// outweigh the whole window, and dropping it would strip the fork of the +/// context it was forked for. pub fn precompact_forked_history( history: &mut Vec, context_window: usize, fraction: f32, ) -> usize { + let budget = forked_history_budget(context_window); let mut removed_total = 0usize; // Re-estimate after each drain: one fractional cut isn't guaranteed to - // land under budget when a few large messages dominate the history. The - // floor of 4 retained messages bounds the loop; a history that still - // exceeds the budget at the floor is left for overflow recovery. - loop { - let estimated = estimate_history_tokens(history); - let usage = estimated as f32 / context_window.max(1) as f32; - if usage < 0.70 { - break; - } - + // land under budget when a few large messages dominate the history. + while estimate_history_tokens(history) + forked_compaction_marker_tokens(removed_total) > budget + { let total = history.len(); - if total <= 4 { + if total <= FORK_MIN_RETAINED_MESSAGES { break; } let remove_count = ((total as f32 * fraction) as usize) .max(1) - .min(total.saturating_sub(2)); + .min(total - FORK_MIN_RETAINED_MESSAGES); history.drain(..remove_count); removed_total += remove_count; } - if removed_total > 0 { - let marker = format!( - "[Forked context compacted: {removed_total} older messages removed to stay within context limits. \ - Continue with the information available.]" + let retained_tokens = estimate_history_tokens(history); + if retained_tokens + forked_compaction_marker_tokens(removed_total) > budget { + tracing::warn!( + retained_tokens, + budget, + retained_messages = history.len(), + "forked history exceeds its budget at the retention floor; the fork \ + may enter context-overflow recovery" ); - history.insert(0, Message::from(marker)); + } + + if removed_total > 0 { + history.insert(0, Message::from(forked_compaction_marker(removed_total))); } removed_total } @@ -505,13 +544,13 @@ mod tests { #[test] fn precompact_drains_until_under_budget() { // 40 messages of 4000 chars ≈ 40k tokens against a 20k window: a - // single 50% cut leaves ~20k (still ≥70%), so the loop must run + // single 50% cut leaves ~20k, still over budget, so the loop must run // more than once. let mut history: Vec = (0..40).map(|_| text_message(4000)).collect(); let removed = precompact_forked_history(&mut history, 20_000, 0.50); assert!(removed > 20, "one fractional cut is not enough: {removed}"); - let estimated = estimate_history_tokens(&history); - assert!((estimated as f32) < 20_000.0 * 0.70); + // The marker is part of what the fork carries, so it counts too. + assert!(estimate_history_tokens(&history) <= forked_history_budget(20_000)); // Marker inserted once, at the front. let front = match &history[0] { Message::User { content } => format!("{content:?}"), @@ -521,13 +560,79 @@ mod tests { assert!(!format!("{:?}", history[1]).contains("Forked context compacted")); } + #[test] + fn precompact_reserves_prompt_and_response_capacity() { + // A history that fits the raw window but not the budget still gets + // trimmed: the fork needs room for its own prompt and response. + let budget = forked_history_budget(20_000); + assert!(budget < 20_000, "reserve must hold something back"); + + let mut history: Vec = (0..10).map(|_| text_message(7_000)).collect(); + let estimated = estimate_history_tokens(&history); + assert!(estimated < 20_000 && estimated > budget, "{estimated}"); + + let removed = precompact_forked_history(&mut history, 20_000, 0.50); + assert!(removed > 0); + assert!(estimate_history_tokens(&history) <= budget); + } + + #[test] + fn precompact_noop_at_exact_budget() { + // 7 messages of 4000 chars = 7000 tokens, exactly the budget for a + // 10k window. Sitting on the line is not over it. + let mut history: Vec = (0..7).map(|_| text_message(4000)).collect(); + assert_eq!( + estimate_history_tokens(&history), + forked_history_budget(10_000) + ); + + let removed = precompact_forked_history(&mut history, 10_000, 0.50); + assert_eq!(removed, 0); + assert_eq!(history.len(), 7); + } + + #[test] + fn precompact_trims_one_token_over_budget() { + let budget = forked_history_budget(10_000); + let mut history: Vec = (0..7).map(|_| text_message(4000)).collect(); + history.push(text_message(4)); + assert_eq!(estimate_history_tokens(&history), budget + 1); + + let removed = precompact_forked_history(&mut history, 10_000, 0.50); + assert!(removed > 0); + assert!(estimate_history_tokens(&history) <= budget); + } + #[test] fn precompact_stops_at_retention_floor() { // A handful of giant messages that can never fit the budget: the // loop must stop at the floor instead of spinning or emptying. let mut history: Vec = (0..6).map(|_| text_message(100_000)).collect(); let removed = precompact_forked_history(&mut history, 10_000, 0.50); - assert!(history.len() >= 4); + assert_eq!( + history.len(), + FORK_MIN_RETAINED_MESSAGES + 1, + "marker + floor" + ); + assert_eq!(removed, 2); + } + + #[test] + fn precompact_keeps_oversized_recent_message() { + // One recent message larger than the entire window: dropping the rest + // can't bring the fork under budget, and the message itself is the + // context the fork was created for, so it survives. + let mut history: Vec = (0..12).map(|_| text_message(2_000)).collect(); + history.push(text_message(400_000)); + + let removed = precompact_forked_history(&mut history, 10_000, 0.50); + assert_eq!( + history.len(), + FORK_MIN_RETAINED_MESSAGES + 1, + "marker + floor" + ); assert!(removed > 0); + let last = history.last().unwrap(); + assert!(estimate_history_tokens(std::slice::from_ref(last)) > 10_000); } } From 31d0a3e5b8e48466f142f1ec2680b9895fa11af5 Mon Sep 17 00:00:00 2001 From: Jamie Pine Date: Sun, 9 Aug 2026 22:30:04 -0700 Subject: [PATCH 6/6] fix(compactor): budget forks against the prompt they will actually carry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 30% reserve was a guess made before the caller attached anything. precompact_forked_history now takes the prompt size the caller is about to send — worker forks measure the fully rendered preamble (skills and ambient memory included) plus the task, branches measure their system prompt plus the user prompt — and reserves 15% of the window for the response on top. A preamble that outgrows the window collapses the budget to zero; the fork keeps its retention floor and warns rather than being emptied. --- src/agent/branch.rs | 10 ++- src/agent/channel_dispatch.rs | 6 ++ src/agent/compactor.rs | 120 +++++++++++++++++++++++----------- 3 files changed, 95 insertions(+), 41 deletions(-) diff --git a/src/agent/branch.rs b/src/agent/branch.rs index d913fa339..b3053e301 100644 --- a/src/agent/branch.rs +++ b/src/agent/branch.rs @@ -106,7 +106,7 @@ impl Branch { // Pre-flight context check: if the forked history is already large, // compact before we even make the first LLM call. - self.maybe_compact_history(); + self.maybe_compact_history(&prompt); let routing = self.deps.runtime_config.routing.load(); let model_name = self @@ -279,14 +279,18 @@ impl Branch { Ok(conclusion) } - /// Compact history down to the fork's token budget, dropping the oldest + /// Compact history down to what the context window has left once this + /// branch's own preamble and prompt are accounted for, dropping the oldest /// half of the messages at a time until it fits. - fn maybe_compact_history(&mut self) { + fn maybe_compact_history(&mut self, prompt: &str) { let context_window = **self.deps.runtime_config.context_window.load(); + let prompt_tokens = crate::agent::compactor::estimate_text_tokens(&self.system_prompt) + + crate::agent::compactor::estimate_text_tokens(prompt); let removed = crate::agent::compactor::precompact_forked_history( &mut self.history, context_window, 0.50, + prompt_tokens, ); if removed > 0 { tracing::info!( diff --git a/src/agent/channel_dispatch.rs b/src/agent/channel_dispatch.rs index 002a3f556..822d0b169 100644 --- a/src/agent/channel_dispatch.rs +++ b/src/agent/channel_dispatch.rs @@ -748,10 +748,16 @@ async fn spawn_worker_inner( WorkerHistoryMode::Fork => { let mut history = state.history.read().await.clone(); let context_window = **state.deps.runtime_config.context_window.load(); + // The preamble is fully rendered by now — skills and ambient + // memory included — so the fork is budgeted against what the + // worker's first call actually leaves for history. + let prompt_tokens = crate::agent::compactor::estimate_text_tokens(&system_prompt) + + crate::agent::compactor::estimate_text_tokens(task); let removed = crate::agent::compactor::precompact_forked_history( &mut history, context_window, 0.50, + prompt_tokens, ); if removed > 0 { tracing::info!( diff --git a/src/agent/compactor.rs b/src/agent/compactor.rs index 402804b4b..08409a2f1 100644 --- a/src/agent/compactor.rs +++ b/src/agent/compactor.rs @@ -287,21 +287,21 @@ async fn run_compaction( Ok(remove_count) } -/// Share of the context window pre-compaction holds back for the fork's system -/// prompt and the model's response. Forked history is budgeted against what -/// remains, so a fork that fits still has room to make its first call. -const FORK_CONTEXT_RESERVE: f32 = 0.30; +/// Share of the context window pre-compaction holds back for the fork's +/// response. The rest of the window covers the fork's prompt input — system +/// prompt, task, and history. +const FORK_RESPONSE_RESERVE: f32 = 0.15; /// Messages pre-compaction never drops. A fork needs its most recent exchange /// to be able to act at all, so the budget yields to it rather than the /// reverse. const FORK_MIN_RETAINED_MESSAGES: usize = 4; -/// Tokens available to a forked history, once the fork's own prompt and -/// response capacity is reserved. -fn forked_history_budget(context_window: usize) -> usize { - let reserve = (context_window as f32 * FORK_CONTEXT_RESERVE) as usize; - context_window.saturating_sub(reserve) +/// Tokens available to a forked history, once the prompt the caller is about +/// to attach and room for a response are both reserved. +fn forked_history_budget(context_window: usize, prompt_tokens: usize) -> usize { + let response_reserve = (context_window as f32 * FORK_RESPONSE_RESERVE) as usize; + context_window.saturating_sub(response_reserve.saturating_add(prompt_tokens)) } fn forked_compaction_marker(removed_total: usize) -> String { @@ -326,6 +326,12 @@ fn forked_compaction_marker_tokens(removed_total: usize) -> usize { /// their first LLM call, so a large parent history doesn't start the fork's /// life in overflow recovery. Returns the number of messages removed. /// +/// `prompt_tokens` is what the caller is about to attach alongside this +/// history — its rendered system prompt (skills and ambient memory included) +/// and the task or user message. Callers measure it with +/// [`estimate_text_tokens`], so the budget reflects the real first call rather +/// than an assumed prompt size. +/// /// The most recent [`FORK_MIN_RETAINED_MESSAGES`] messages are preserved even /// when they alone exceed the budget — a single oversized tool result can /// outweigh the whole window, and dropping it would strip the fork of the @@ -334,8 +340,9 @@ pub fn precompact_forked_history( history: &mut Vec, context_window: usize, fraction: f32, + prompt_tokens: usize, ) -> usize { - let budget = forked_history_budget(context_window); + let budget = forked_history_budget(context_window, prompt_tokens); let mut removed_total = 0usize; // Re-estimate after each drain: one fractional cut isn't guaranteed to @@ -359,6 +366,7 @@ pub fn precompact_forked_history( tracing::warn!( retained_tokens, budget, + prompt_tokens, retained_messages = history.len(), "forked history exceeds its budget at the retention floor; the fork \ may enter context-overflow recovery" @@ -371,6 +379,13 @@ pub fn precompact_forked_history( removed_total } +/// Estimate the token count of prompt text with the same chars/4 heuristic +/// [`estimate_history_tokens`] uses. Rounds up, so a budget built on it never +/// understates what the prompt will occupy. +pub fn estimate_text_tokens(text: &str) -> usize { + text.len().div_ceil(4) +} + /// Estimate token count for a history using chars/4 heuristic. /// /// This is intentionally rough — it's only used for threshold checks, not billing. @@ -533,24 +548,31 @@ mod tests { Message::from("x".repeat(size)) } + #[test] + fn estimate_text_tokens_rounds_up() { + assert_eq!(estimate_text_tokens(""), 0); + assert_eq!(estimate_text_tokens("abcd"), 1); + assert_eq!(estimate_text_tokens("abcde"), 2); + } + #[test] fn precompact_noop_under_budget() { let mut history: Vec = (0..10).map(|_| text_message(100)).collect(); - let removed = precompact_forked_history(&mut history, 200_000, 0.50); + let removed = precompact_forked_history(&mut history, 200_000, 0.50, 0); assert_eq!(removed, 0); assert_eq!(history.len(), 10); } #[test] fn precompact_drains_until_under_budget() { - // 40 messages of 4000 chars ≈ 40k tokens against a 20k window: a - // single 50% cut leaves ~20k, still over budget, so the loop must run + // 40 messages of 4000 chars = 40k tokens against a 20k window: a + // single 50% cut leaves 20k, still over budget, so the loop must run // more than once. let mut history: Vec = (0..40).map(|_| text_message(4000)).collect(); - let removed = precompact_forked_history(&mut history, 20_000, 0.50); + let removed = precompact_forked_history(&mut history, 20_000, 0.50, 0); assert!(removed > 20, "one fractional cut is not enough: {removed}"); // The marker is part of what the fork carries, so it counts too. - assert!(estimate_history_tokens(&history) <= forked_history_budget(20_000)); + assert!(estimate_history_tokens(&history) <= forked_history_budget(20_000, 0)); // Marker inserted once, at the front. let front = match &history[0] { Message::User { content } => format!("{content:?}"), @@ -561,44 +583,50 @@ mod tests { } #[test] - fn precompact_reserves_prompt_and_response_capacity() { - // A history that fits the raw window but not the budget still gets - // trimmed: the fork needs room for its own prompt and response. - let budget = forked_history_budget(20_000); - assert!(budget < 20_000, "reserve must hold something back"); - - let mut history: Vec = (0..10).map(|_| text_message(7_000)).collect(); - let estimated = estimate_history_tokens(&history); - assert!(estimated < 20_000 && estimated > budget, "{estimated}"); + fn precompact_budget_tracks_the_measured_prompt() { + // 15k tokens of history in a 20k window. It fits once only the + // response is reserved, and stops fitting when the caller reports a + // 4k-token preamble — the same history, budgeted against the real + // first call. + let history: Vec = (0..10).map(|_| text_message(6_000)).collect(); + assert_eq!(estimate_history_tokens(&history), 15_000); + + let mut without_prompt = history.clone(); + assert_eq!( + precompact_forked_history(&mut without_prompt, 20_000, 0.50, 0), + 0 + ); - let removed = precompact_forked_history(&mut history, 20_000, 0.50); - assert!(removed > 0); - assert!(estimate_history_tokens(&history) <= budget); + let mut with_prompt = history; + let removed = precompact_forked_history(&mut with_prompt, 20_000, 0.50, 4_000); + assert!(removed > 0, "a 4k preamble must push this history over"); + assert!(estimate_history_tokens(&with_prompt) <= forked_history_budget(20_000, 4_000)); } #[test] fn precompact_noop_at_exact_budget() { - // 7 messages of 4000 chars = 7000 tokens, exactly the budget for a - // 10k window. Sitting on the line is not over it. - let mut history: Vec = (0..7).map(|_| text_message(4000)).collect(); + // 8 messages of 4000 chars = 8000 tokens, exactly the budget for a + // 10k window with a 500-token prompt. Sitting on the line is not over + // it. + let mut history: Vec = (0..8).map(|_| text_message(4000)).collect(); assert_eq!( estimate_history_tokens(&history), - forked_history_budget(10_000) + forked_history_budget(10_000, 500) ); - let removed = precompact_forked_history(&mut history, 10_000, 0.50); + let removed = precompact_forked_history(&mut history, 10_000, 0.50, 500); assert_eq!(removed, 0); - assert_eq!(history.len(), 7); + assert_eq!(history.len(), 8); } #[test] fn precompact_trims_one_token_over_budget() { - let budget = forked_history_budget(10_000); - let mut history: Vec = (0..7).map(|_| text_message(4000)).collect(); + let budget = forked_history_budget(10_000, 500); + let mut history: Vec = (0..8).map(|_| text_message(4000)).collect(); history.push(text_message(4)); assert_eq!(estimate_history_tokens(&history), budget + 1); - let removed = precompact_forked_history(&mut history, 10_000, 0.50); + let removed = precompact_forked_history(&mut history, 10_000, 0.50, 500); assert!(removed > 0); assert!(estimate_history_tokens(&history) <= budget); } @@ -608,7 +636,7 @@ mod tests { // A handful of giant messages that can never fit the budget: the // loop must stop at the floor instead of spinning or emptying. let mut history: Vec = (0..6).map(|_| text_message(100_000)).collect(); - let removed = precompact_forked_history(&mut history, 10_000, 0.50); + let removed = precompact_forked_history(&mut history, 10_000, 0.50, 0); assert_eq!( history.len(), FORK_MIN_RETAINED_MESSAGES + 1, @@ -625,7 +653,7 @@ mod tests { let mut history: Vec = (0..12).map(|_| text_message(2_000)).collect(); history.push(text_message(400_000)); - let removed = precompact_forked_history(&mut history, 10_000, 0.50); + let removed = precompact_forked_history(&mut history, 10_000, 0.50, 0); assert_eq!( history.len(), FORK_MIN_RETAINED_MESSAGES + 1, @@ -635,4 +663,20 @@ mod tests { let last = history.last().unwrap(); assert!(estimate_history_tokens(std::slice::from_ref(last)) > 10_000); } + + #[test] + fn precompact_handles_prompt_larger_than_the_window() { + // A preamble that alone outgrows the window leaves no budget at all. + // The fork still keeps its floor rather than being emptied. + assert_eq!(forked_history_budget(10_000, 20_000), 0); + + let mut history: Vec = (0..10).map(|_| text_message(1_000)).collect(); + let removed = precompact_forked_history(&mut history, 10_000, 0.50, 20_000); + assert_eq!(removed, 6); + assert_eq!( + history.len(), + FORK_MIN_RETAINED_MESSAGES + 1, + "marker + floor" + ); + } }