diff --git a/src/app.rs b/src/app.rs index 3471c9c..891b3e1 100644 --- a/src/app.rs +++ b/src/app.rs @@ -1408,8 +1408,10 @@ impl App { let name = s.name.clone(); let kind = s.kind.label().to_string(); let waiting_prompt = s.waiting_prompt.clone(); - let tail = pipe::extract_from_session(&self.sessions, session_id, &ExtractMode::LastN(15)) - .unwrap_or_default(); + let tail_n = self.config.orchestrator.event_tail_lines.max(1); + let tail = + pipe::extract_from_session(&self.sessions, session_id, &ExtractMode::LastN(tail_n)) + .unwrap_or_default(); if let Some(h) = &self.orchestrator { // try_send: a wedged orchestrator must never block the main loop. @@ -2061,6 +2063,20 @@ impl App { } }) .unwrap_or_default(); + // Cap the reply so a chatty session can't dump its whole + // transcript into the orchestrator's context in one tool result. + let max = self.config.orchestrator.wait_ready_max_lines; + let lines = if max > 0 && lines.len() > max { + let total = lines.len(); + let mut kept: Vec = lines[total - max..].to_vec(); + kept.insert( + 0, + format!("[truncated: showing last {} of {} lines]", max, total), + ); + kept + } else { + lines + }; let _ = resp_tx.send(serde_json::json!({ "session_id": session_id, "lines": lines, diff --git a/src/config.rs b/src/config.rs index cd083bc..52d1c0b 100644 --- a/src/config.rs +++ b/src/config.rs @@ -101,6 +101,21 @@ pub struct OrchestratorConfig { pub max_tokens: u32, pub max_tool_iterations: usize, pub input_wait_timeout_secs: u64, + /// Soft token budget for the conversation history, estimated at ~4 + /// chars/token. When the estimate exceeds this, oldest turns are dropped + /// even if max_history_turns hasn't been reached. 0 disables. + pub max_context_tokens: usize, + /// How many recent user turns keep their tool results verbatim. Tool + /// results older than this are replaced with a short elision stub (the + /// model can re-run the tool if it still needs the data). 0 disables. + pub tool_result_keep_turns: usize, + /// Lines of session output inlined into a [linkshell event] + /// notification. The orchestrator can always read_output for more. + pub event_tail_lines: usize, + /// Cap on lines returned by send_input wait_ready / `input --wait`. + /// Longer replies are truncated to the last N lines with a marker. + /// 0 disables. + pub wait_ready_max_lines: usize, } impl Default for OrchestratorConfig { @@ -139,6 +154,10 @@ impl Default for OrchestratorConfig { max_tokens: 4096, max_tool_iterations: 12, input_wait_timeout_secs: 180, + max_context_tokens: 60_000, + tool_result_keep_turns: 3, + event_tail_lines: 5, + wait_ready_max_lines: 80, } } } diff --git a/src/orchestrator/anthropic.rs b/src/orchestrator/anthropic.rs index 8e2cc32..88862a6 100644 --- a/src/orchestrator/anthropic.rs +++ b/src/orchestrator/anthropic.rs @@ -4,6 +4,32 @@ use crate::config::OrchestratorConfig; use crate::events::AppEvent; use tokio::sync::mpsc; +/// Clone the history for one request, attaching an ephemeral cache_control +/// breakpoint to the final content block of the last message. The stored +/// history is left untouched (trim_history relies on plain-string user +/// turns), and the last message at request time is always a user message +/// (fresh user text or tool_results), both of which accept cache_control. +fn messages_with_cache_breakpoint(history: &[serde_json::Value]) -> serde_json::Value { + let mut msgs: Vec = history.to_vec(); + if let Some(last) = msgs.last_mut() { + let ce = serde_json::json!({"type": "ephemeral"}); + let content = last["content"].take(); + last["content"] = match content { + serde_json::Value::String(text) => { + serde_json::json!([{"type": "text", "text": text, "cache_control": ce}]) + } + serde_json::Value::Array(mut blocks) => { + if let Some(b) = blocks.last_mut() { + b["cache_control"] = ce; + } + serde_json::Value::Array(blocks) + } + other => other, + }; + } + serde_json::Value::Array(msgs) +} + /// Run one conversation turn: append the user text, loop through tool calls, /// return the final assistant text. History uses the Messages API shape and /// assistant content (including thinking blocks) is replayed verbatim. @@ -22,11 +48,22 @@ pub async fn run_turn( ) })?; let url = format!("{}/v1/messages", cfg.endpoint_url().trim_end_matches('/')); - let tools = super::anthropic_tools(); - let system = super::system_prompt(cfg); + // Prompt caching: mark the static prefix (tools + system) once, and put + // a moving breakpoint on the last message of each request so the 12 + // iterations of a busy turn re-serve the shared history prefix instead + // of re-billing it in full every call. + let mut tools = super::anthropic_tools(); + if let Some(last) = tools.as_array_mut().and_then(|a| a.last_mut()) { + last["cache_control"] = serde_json::json!({"type": "ephemeral"}); + } + let system = serde_json::json!([{ + "type": "text", + "text": super::system_prompt(cfg), + "cache_control": {"type": "ephemeral"} + }]); history.push(serde_json::json!({"role": "user", "content": user_text})); - super::trim_history(history, cfg.max_history_turns); + super::compact_history(history, cfg); for i in 0..cfg.max_tool_iterations { if interrupt.hit() { @@ -42,7 +79,7 @@ pub async fn run_turn( "max_tokens": cfg.max_tokens, "system": system, "tools": tools, - "messages": history, + "messages": messages_with_cache_breakpoint(history), }); let resp: serde_json::Value = auth .apply(client.post(&url)) @@ -135,7 +172,7 @@ pub async fn run_turn( "system": system, "tools": tools, "tool_choice": {"type": "none"}, - "messages": history, + "messages": messages_with_cache_breakpoint(history), }); let resp: serde_json::Value = auth .apply(client.post(&url)) diff --git a/src/orchestrator/mod.rs b/src/orchestrator/mod.rs index b838c17..0aeca83 100644 --- a/src/orchestrator/mod.rs +++ b/src/orchestrator/mod.rs @@ -783,6 +783,98 @@ fn skills_section(cfg: &OrchestratorConfig, with_paths: bool) -> Option Some(skills::skill_list(&list, with_paths)) } +/// Stub text substituted for aged-out tool results. +const ELIDED_RESULT: &str = + "[elided to save context — re-run the tool if this result is still needed]"; + +/// Count plain user-text turns (the boundaries trim_history cuts at). +fn user_turns(history: &[serde_json::Value]) -> usize { + history + .iter() + .filter(|m| m["role"] == "user" && m["content"].is_string()) + .count() +} + +/// Rough token estimate for the serialized history (~4 chars/token). +fn estimate_tokens(history: &[serde_json::Value]) -> usize { + history + .iter() + .map(|m| m.to_string().chars().count()) + .sum::() + / 4 +} + +/// Replace tool results older than the last `keep_turns` plain user turns +/// with a short stub. Structure stays API-valid for both providers: the +/// anthropic shape keeps its tool_result blocks (ids intact) with stubbed +/// content, the openai shape keeps its role:"tool" messages likewise. +fn age_tool_results(history: &mut [serde_json::Value], keep_turns: usize) { + if keep_turns == 0 { + return; + } + // Index of the keep_turns-th plain user turn from the end; everything + // before it is "old". + let mut seen = 0; + let mut boundary = 0; + for (i, m) in history.iter().enumerate().rev() { + if m["role"] == "user" && m["content"].is_string() { + seen += 1; + if seen == keep_turns { + boundary = i; + break; + } + } + } + if seen < keep_turns { + return; // whole history is within the keep window + } + for m in history[..boundary].iter_mut() { + // OpenAI shape: {"role": "tool", "content": "..."} + if m["role"] == "tool" { + if m["content"] + .as_str() + .is_some_and(|s| s.len() > ELIDED_RESULT.len()) + { + m["content"] = serde_json::json!(ELIDED_RESULT); + } + continue; + } + // Anthropic shape: user message whose content is an array of + // tool_result blocks. + if m["role"] == "user" { + if let Some(blocks) = m["content"].as_array_mut() { + for b in blocks.iter_mut().filter(|b| b["type"] == "tool_result") { + let long = match &b["content"] { + serde_json::Value::String(s) => s.len() > ELIDED_RESULT.len(), + v => v.to_string().len() > ELIDED_RESULT.len(), + }; + if long { + b["content"] = serde_json::json!(ELIDED_RESULT); + } + } + } + } + } +} + +/// Full history compaction pass, run once per turn before hitting the API: +/// age old tool results, apply the turn cap, then drop oldest turns until +/// the token estimate fits the budget (always keeping the latest turn). +pub(crate) fn compact_history(history: &mut Vec, cfg: &OrchestratorConfig) { + age_tool_results(history, cfg.tool_result_keep_turns); + trim_history(history, cfg.max_history_turns); + if cfg.max_context_tokens == 0 { + return; + } + while estimate_tokens(history) > cfg.max_context_tokens { + let turns = user_turns(history); + if turns <= 1 { + break; // never drop the turn we're about to answer + } + trim_history(history, turns - 1); + } +} + /// Trim provider history in place, dropping oldest turns but only cutting at /// plain user-text boundaries so tool_use/tool_result pairs stay intact. fn trim_history(history: &mut Vec, max_turns: usize) { @@ -1009,6 +1101,79 @@ mod tests { assert_eq!(history.len(), 1); } + #[test] + fn aging_stubs_only_old_tool_results() { + let big = "x".repeat(500); + let mut h = vec![ + serde_json::json!({"role": "user", "content": "one"}), + serde_json::json!({"role": "assistant", "content": [{"type": "tool_use", "id": "a"}]}), + serde_json::json!({"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "a", "content": big.clone()}]}), + serde_json::json!({"role": "assistant", "content": "reply"}), + serde_json::json!({"role": "user", "content": "two"}), + serde_json::json!({"role": "assistant", "content": [{"type": "tool_use", "id": "b"}]}), + serde_json::json!({"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "b", "content": big.clone()}]}), + ]; + age_tool_results(&mut h, 1); + // Old result stubbed, id preserved + assert_eq!(h[2]["content"][0]["content"], ELIDED_RESULT); + assert_eq!(h[2]["content"][0]["tool_use_id"], "a"); + // Result within the keep window untouched + assert_eq!(h[6]["content"][0]["content"], big); + // keep_turns larger than history: no-op + let before = h.clone(); + age_tool_results(&mut h, 10); + assert_eq!(h, before); + } + + #[test] + fn aging_stubs_openai_tool_messages() { + let big = "y".repeat(500); + let mut h = vec![ + serde_json::json!({"role": "user", "content": "one"}), + serde_json::json!({"role": "tool", "content": big}), + serde_json::json!({"role": "user", "content": "two"}), + ]; + age_tool_results(&mut h, 1); + assert_eq!(h[1]["content"], ELIDED_RESULT); + } + + #[test] + fn budget_trim_drops_oldest_turns_but_keeps_the_last() { + let big = "z".repeat(4000); // ~1000 tokens per turn + let mut h: Vec = (0..10) + .flat_map(|i| { + vec![ + serde_json::json!({"role": "user", "content": format!("{} {}", i, big)}), + serde_json::json!({"role": "assistant", "content": "ok"}), + ] + }) + .collect(); + let cfg = OrchestratorConfig { + max_history_turns: 40, + max_context_tokens: 3000, + tool_result_keep_turns: 0, + ..Default::default() + }; + compact_history(&mut h, &cfg); + let turns = user_turns(&h); + assert!(turns < 10, "should have dropped turns, kept {}", turns); + assert!(turns >= 1, "must keep at least the latest turn"); + // Latest turn survives + assert!(h + .iter() + .any(|m| m["content"].as_str().is_some_and(|s| s.starts_with("9 ")))); + // Budget disabled: only the turn cap applies + let mut h2 = vec![serde_json::json!({"role": "user", "content": "u".repeat(100000)})]; + let cfg2 = OrchestratorConfig { + max_context_tokens: 0, + ..Default::default() + }; + compact_history(&mut h2, &cfg2); + assert_eq!(h2.len(), 1); + } + #[test] fn trim_history_cuts_only_at_plain_user_turns() { // 3 user turns, with an assistant tool_use + user tool_result between diff --git a/src/orchestrator/openai.rs b/src/orchestrator/openai.rs index c8ed602..373072c 100644 --- a/src/orchestrator/openai.rs +++ b/src/orchestrator/openai.rs @@ -21,7 +21,7 @@ pub async fn run_turn( let tools = super::openai_tools(); history.push(serde_json::json!({"role": "user", "content": user_text})); - super::trim_history(history, cfg.max_history_turns); + super::compact_history(history, cfg); for i in 0..cfg.max_tool_iterations { if interrupt.hit() {