From d1ae121e62ca24a8237e1d254b34caef38819742 Mon Sep 17 00:00:00 2001 From: Yanuar Date: Tue, 13 Oct 2026 10:31:08 +0700 Subject: [PATCH 1/2] fix(acp): compact using provider context usage --- src/acp/service.rs | 599 ++++++++++++++++++++++++++++++++++---- src/session/compaction.rs | 10 +- tests/acp_stdio.rs | 307 ++++++++++++++++++- 3 files changed, 851 insertions(+), 65 deletions(-) diff --git a/src/acp/service.rs b/src/acp/service.rs index 11a35eb..97f3380 100644 --- a/src/acp/service.rs +++ b/src/acp/service.rs @@ -1429,42 +1429,6 @@ impl AcpService { user_message.provider = Some(session.provider.clone()); user_message.model = Some(session.model.clone()); user_message.agent_mode = Some(session.agent.clone()); - let auto_compacted = self - .maybe_auto_compact_session( - &session_id, - &session, - cancellation.clone(), - crate::session::compaction::message_context_tokens(&user_message), - ) - .await?; - if cancellation.is_cancelled() { - return Ok(PromptResponse::new(StopReason::Cancelled)); - } - if auto_compacted { - messages = self - .session_manager - .lock() - .map_err(|_| internal_error())? - .get_session_ref(&session_id) - .map(|stored| stored.messages.clone()) - .ok_or_else(|| Error::invalid_params().data("unknown session"))?; - } - { - let mut manager = self.session_manager.lock().map_err(|_| internal_error())?; - manager - .add_message_to_session(&session_id, &user_message) - .map_err(|_| internal_error())?; - attachment_guard.commit(); - manager - .set_session_status( - &session_id, - crate::session::types::SessionStatus::Streaming, - None, - ) - .map_err(|_| internal_error())?; - } - messages.push(user_message); - let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); let process_registry = std::sync::Arc::new(crate::tools::ProcessRegistry::new()); let prompt_registry = cancellable( @@ -1504,12 +1468,48 @@ impl AcpService { let Some(system_prompt) = system_prompt else { return Ok(PromptResponse::new(StopReason::Cancelled)); }; - messages.insert(0, crate::session::types::Message::system(system_prompt)); - let base_context_tokens = crate::session::compaction::total_context_tokens(&messages); - let base_cost = messages - .iter() - .filter_map(|message| message.cost) - .sum::(); + let system_message = crate::session::types::Message::system(system_prompt); + let prompt_tokens = crate::session::compaction::message_context_tokens(&system_message); + let auto_compacted = self + .maybe_auto_compact_session( + &session_id, + &session, + &connection, + cancellation.clone(), + prompt_tokens, + crate::session::compaction::message_context_tokens(&user_message), + ) + .await?; + if cancellation.is_cancelled() { + return Ok(PromptResponse::new(StopReason::Cancelled)); + } + if auto_compacted { + messages = self + .session_manager + .lock() + .map_err(|_| internal_error())? + .get_session_ref(&session_id) + .map(|stored| stored.messages.clone()) + .ok_or_else(|| Error::invalid_params().data("unknown session"))?; + } + { + let mut manager = self.session_manager.lock().map_err(|_| internal_error())?; + manager + .add_message_to_session(&session_id, &user_message) + .map_err(|_| internal_error())?; + attachment_guard.commit(); + manager + .set_session_status( + &session_id, + crate::session::types::SessionStatus::Streaming, + None, + ) + .map_err(|_| internal_error())?; + } + messages.push(user_message); + let base_context_tokens = acp_context_tokens(&messages, prompt_tokens); + let base_cost = acp_session_cost(&messages); + messages = acp_request_messages(&messages, system_message); send_usage( &connection, &session_id, @@ -1565,12 +1565,20 @@ impl AcpService { &cancellation, receiver, base_context_tokens, + prompt_tokens, base_cost, ) .await?; if !cancellation.is_cancelled() && response.stop_reason != StopReason::Cancelled { let _ = self - .maybe_auto_compact_session(&session_id, &session, cancellation.clone(), 0) + .maybe_auto_compact_session( + &session_id, + &session, + &connection, + cancellation.clone(), + prompt_tokens, + 0, + ) .await?; } Ok(response) @@ -1585,6 +1593,7 @@ impl AcpService { cancellation: &CancellationToken, mut receiver: tokio::sync::mpsc::UnboundedReceiver, base_context_tokens: usize, + base_prompt_tokens: usize, base_cost: f64, ) -> Result { let mut assistant = crate::session::types::Message::incomplete(""); @@ -1693,6 +1702,9 @@ impl AcpService { usage.cache_write, estimate_session_usage_cost(session, &usage), )); + if let Some(part) = assistant.parts.last_mut() { + part.data["acp_prompt_tokens"] = serde_json::json!(base_prompt_tokens); + } if usage.output > 0 { assistant.output_tokens = Some( assistant @@ -1840,17 +1852,7 @@ impl AcpService { match result { Ok(stats) => { - let feedback = format!( - "Context compacted ({})", - crate::session::compaction::format_compaction_stats(stats) - ); - send_text( - &connection, - session_id, - &cuid2::create_id(), - feedback, - false, - )?; + self.notify_compaction(session_id, &session, &connection, stats, 0)?; Ok(PromptResponse::new(StopReason::EndTurn)) } Err(_error) if cancellation.is_cancelled() => { @@ -1959,7 +1961,9 @@ impl AcpService { &self, session_id: &str, session: &AcpSession, + connection: &ConnectionTo, cancellation: CancellationToken, + prompt_tokens: usize, additional_tokens: usize, ) -> Result { let messages = self @@ -1969,8 +1973,8 @@ impl AcpService { .get_session_ref(session_id) .map(|stored| stored.messages.clone()) .ok_or_else(|| Error::invalid_params().data("unknown session"))?; - let used_tokens = crate::session::compaction::total_context_tokens(&messages) - .saturating_add(additional_tokens); + let used_tokens = + acp_context_tokens(&messages, prompt_tokens).saturating_add(additional_tokens); if !crate::session::compaction::should_auto_compact( &session.config.merged_config.compaction, used_tokens, @@ -2018,13 +2022,125 @@ impl AcpService { ) .map_err(|_| internal_error())?; match result { - Ok(_) => Ok(true), + Ok(stats) => { + self.notify_compaction(session_id, session, connection, stats, prompt_tokens)?; + Ok(true) + } Err(error) => { crate::emit_log!("ACP auto-compaction skipped after failure: {:?}", error); Ok(false) } } } + + fn notify_compaction( + &self, + session_id: &str, + session: &AcpSession, + connection: &ConnectionTo, + stats: crate::session::types::CompactionStats, + prompt_tokens: usize, + ) -> Result<(), Error> { + let (used_tokens, cost) = { + let manager = self.session_manager.lock().map_err(|_| internal_error())?; + let stored = manager + .get_session_ref(session_id) + .ok_or_else(|| Error::invalid_params().data("unknown session"))?; + ( + acp_context_tokens(&stored.messages, prompt_tokens), + acp_session_cost(&stored.messages), + ) + }; + send_text( + connection, + session_id, + &cuid2::create_id(), + format!( + "Context compacted ({})", + crate::session::compaction::format_compaction_stats(stats) + ), + false, + )?; + send_usage( + connection, + session_id, + session, + used_tokens, + (cost > 0.0).then_some(cost), + None, + ) + } +} + +fn acp_session_cost(messages: &[crate::session::types::Message]) -> f64 { + messages + .iter() + .map(|message| message.cost.unwrap_or_else(|| message.usage_cost())) + .sum() +} + +fn acp_request_messages( + history: &[crate::session::types::Message], + system_message: crate::session::types::Message, +) -> Vec { + let mut messages = crate::session::compaction::filter_messages_for_context(history); + messages.insert(0, system_message); + messages +} + +fn acp_context_tokens(messages: &[crate::session::types::Message], prompt_tokens: usize) -> usize { + let estimated = + crate::session::compaction::total_context_tokens(messages).saturating_add(prompt_tokens); + // Retained tail usage still describes the old context, as does the summary's billed usage. + let start = messages + .iter() + .rposition(crate::session::compaction::is_compaction_marker) + .map(|index| index + 1) + .unwrap_or(0); + for (index, message) in messages.iter().enumerate().skip(start).rev() { + if message.role != crate::session::types::MessageRole::Assistant { + continue; + } + // Each tool step bills its prompt again; cumulative usage is not a context size. + let provider_context = message + .parts + .iter() + .enumerate() + .filter_map(|(part_index, part)| { + let usage = part.recorded_usage()?; + let input = usage + .input + .saturating_add(usage.cache_read) + .saturating_add(usage.cache_write); + if input == 0 { + return None; + } + let mut tokens = + usize::try_from(input.saturating_add(usage.output)).unwrap_or(usize::MAX); + if let Some(previous_prompt) = part + .data + .get("acp_prompt_tokens") + .and_then(serde_json::Value::as_u64) + { + tokens = tokens + .saturating_sub(usize::try_from(previous_prompt).unwrap_or(usize::MAX)) + .saturating_add(prompt_tokens); + } + Some(tokens.saturating_add( + crate::session::compaction::assistant_parts_context_tokens( + &message.parts[part_index + 1..], + ), + )) + }) + .max() + .filter(|tokens| *tokens > 0); + if let Some(tokens) = provider_context { + let following = + crate::session::compaction::total_context_tokens(&messages[index + 1..]); + return estimated.max(tokens.saturating_add(following)); + } + } + estimated } fn compaction_reasoning(session: &AcpSession) -> Option { @@ -3529,6 +3645,373 @@ mod tests { assert_eq!(live.cumulative.cache_read, 60_000); } + #[test] + fn acp_compaction_counts_provider_context_and_cached_input() { + use crate::session::types::{Message, MessagePart}; + let mut assistant = Message::assistant("answer"); + assistant + .parts + .push(MessagePart::usage(60_000, 200, 20_000, 5_000, 0.0)); + let messages = vec![Message::user("question"), assistant]; + + assert_eq!(acp_context_tokens(&messages, 1_000), 85_200); + } + + #[test] + fn acp_compaction_does_not_sum_context_across_tool_requests() { + use crate::session::types::{Message, MessagePart}; + let mut assistant = Message::assistant("answer"); + assistant + .parts + .push(MessagePart::usage(60_000, 200, 20_000, 0, 0.0)); + assistant + .parts + .push(MessagePart::usage(10_000, 100, 40_000, 0, 0.0)); + + assert_eq!( + acp_context_tokens(&[Message::user("question"), assistant], 0), + 80_200 + ); + } + + #[test] + fn acp_compaction_uses_latest_turn_usage_and_counts_pending_history() { + use crate::session::types::{Message, MessagePart}; + let mut old = Message::assistant("old answer"); + old.parts.push(MessagePart::usage(90_000, 100, 0, 0, 0.0)); + let mut latest = Message::assistant("new answer"); + latest + .parts + .push(MessagePart::usage(50_000, 200, 0, 0, 0.0)); + let pending = Message::user("continue"); + let pending_tokens = crate::session::compaction::message_context_tokens(&pending); + let messages = vec![Message::user("question"), old, latest, pending]; + + assert_eq!(acp_context_tokens(&messages, 0), 50_200 + pending_tokens); + } + + #[test] + fn acp_compaction_discards_provider_usage_before_compaction_boundary() { + use crate::session::types::{CompactionStats, Message, MessagePart}; + let mut assistant = Message::assistant("old answer"); + assistant + .parts + .push(MessagePart::usage(90_000, 100, 0, 0, 0.0)); + let mut summary = Message::user(format!( + "{}\nsummary", + crate::session::compaction::SUMMARY_PREFIX + )); + summary + .parts + .push(MessagePart::usage(95_000, 500, 0, 0, 0.0)); + let marker = crate::session::compaction::compaction_marker(CompactionStats { + before_tokens: 90_100, + after_tokens: 100, + before_messages: 2, + after_messages: 1, + }); + let messages = vec![Message::user("question"), assistant, summary, marker]; + + assert_eq!( + acp_context_tokens(&messages, 1_000), + crate::session::compaction::total_context_tokens(&messages) + 1_000 + ); + } + + #[test] + fn acp_compaction_honors_reserved_tokens_and_auto_setting() { + use crate::config::configuration::CompactionConfig; + use crate::session::types::{Message, MessagePart}; + let mut assistant = Message::assistant("answer"); + assistant + .parts + .push(MessagePart::usage(70_000, 0, 10_000, 0, 0.0)); + let messages = vec![Message::user("history ".repeat(2_000)), assistant]; + assert!(crate::session::compaction::select_messages_for_compaction( + &messages, + crate::session::compaction::DEFAULT_TAIL_TURNS, + ) + .is_some()); + let config = CompactionConfig::Settings { + auto: true, + prune: false, + reserved: Some(20_000), + }; + let used = acp_context_tokens(&messages, 1_000); + assert!(crate::session::compaction::total_context_tokens(&messages) < 80_000); + assert!(crate::session::compaction::should_auto_compact( + &config, + used, + Some(100_000), + None + )); + assert!(!crate::session::compaction::should_auto_compact( + &config, + used, + Some(100_001), + None + )); + assert!(!crate::session::compaction::should_auto_compact( + &CompactionConfig::Disabled, + used, + Some(100_000), + None + )); + assert!(!crate::session::compaction::should_auto_compact( + &config, used, None, None + )); + let no_auto = CompactionConfig::Settings { + auto: false, + prune: false, + reserved: Some(20_000), + }; + assert!(!crate::session::compaction::should_auto_compact( + &no_auto, + used, + Some(100_000), + None + )); + } + + #[test] + fn acp_compaction_counts_prompt_and_pending_tokens_without_provider_usage() { + use crate::config::configuration::CompactionConfig; + use crate::session::types::Message; + let messages = vec![Message::user("history ".repeat(2_000))]; + let history_tokens = crate::session::compaction::total_context_tokens(&messages); + let config = CompactionConfig::Settings { + auto: true, + prune: false, + reserved: Some(1_000), + }; + let context_window = Some(history_tokens as u32 + 3_000); + assert!(!crate::session::compaction::should_auto_compact( + &config, + acp_context_tokens(&messages, 1_500), + context_window, + None, + )); + assert!(crate::session::compaction::should_auto_compact( + &config, + acp_context_tokens(&messages, 1_500) + 500, + context_window, + None, + )); + } + + #[test] + fn acp_compaction_replaces_previous_prompt_estimate_in_provider_context() { + use crate::session::types::{Message, MessagePart}; + let mut assistant = Message::assistant("answer"); + let mut usage = MessagePart::usage(80_000, 200, 0, 0, 0.0); + usage.data["acp_prompt_tokens"] = serde_json::json!(5_000); + assistant.parts.push(usage); + let messages = vec![Message::user("question"), assistant]; + assert_eq!(acp_context_tokens(&messages, 25_000), 100_200); + assert_eq!(acp_context_tokens(&messages, 1_000), 76_200); + } + + #[test] + fn acp_compaction_keeps_provider_context_when_new_input_usage_is_missing() { + use crate::session::types::{Message, MessagePart}; + for output in [0, 100] { + let mut previous = Message::assistant("previous answer"); + previous + .parts + .push(MessagePart::usage(85_000, 200, 0, 0, 0.0)); + let mut latest = Message::assistant("new answer"); + let mut usage = MessagePart::usage(0, output, 0, 0, 0.0); + usage.data["acp_prompt_tokens"] = serde_json::json!(1_000); + latest.parts.push(usage); + let following_tokens = crate::session::compaction::message_context_tokens(&latest); + assert_eq!( + acp_context_tokens(&[Message::user("question"), previous, latest], 2_000), + 85_200 + following_tokens + ); + } + } + + #[test] + fn acp_compaction_counts_tool_output_after_provider_usage() { + use crate::session::types::{Message, MessagePart}; + let mut assistant = Message::assistant("answer"); + assistant + .parts + .push(MessagePart::usage(80_000, 200, 0, 0, 0.0)); + record_tool_result( + &mut assistant, + &crate::llm::ToolCallResult { + tool_call_id: "call_1".into(), + role: "tool".into(), + name: "read".into(), + content: serde_json::json!({"output_preview": "late tool output ".repeat(1_000)}) + .to_string(), + }, + ); + let mut following = Message::assistant(""); + following.parts = assistant.parts[2..].to_vec(); + let extra_tokens = crate::session::compaction::message_context_tokens(&following); + assert!(extra_tokens > 1_000); + assert_eq!( + acp_context_tokens(&[Message::user("question"), assistant], 0), + 80_200 + extra_tokens + ); + } + + #[test] + fn acp_compaction_keeps_fresh_system_prompt_in_request() { + use crate::session::types::{CompactionStats, Message}; + let summary = Message::user(format!( + "{}\nsummary", + crate::session::compaction::SUMMARY_PREFIX + )); + let tail = Message::assistant("retained answer"); + let marker = crate::session::compaction::compaction_marker(CompactionStats { + before_tokens: 10_000, + after_tokens: 100, + before_messages: 3, + after_messages: 2, + }); + let pending = Message::user("continue"); + let history = vec![ + Message::user("archived history"), + summary.clone(), + tail.clone(), + marker, + pending.clone(), + ]; + let system = Message::system("fresh system instructions"); + let request = acp_request_messages(&history, system.clone()); + assert_eq!(request, vec![system, summary, tail, pending]); + assert_eq!( + crate::session::compaction::filter_messages_for_context(&request), + request + ); + } + + #[test] + fn acp_compaction_cost_keeps_summary_billing_without_double_counting() { + use crate::session::types::{Message, MessagePart}; + let mut assistant = Message::assistant("answer"); + assistant.cost = Some(0.25); + assistant + .parts + .push(MessagePart::usage(100, 20, 0, 0, 0.25)); + let mut summary = Message::user(format!( + "{}\nsummary", + crate::session::compaction::SUMMARY_PREFIX + )); + summary.parts.push(MessagePart::usage(100, 20, 0, 0, 0.05)); + assert!((acp_session_cost(&[assistant, summary]) - 0.30).abs() < f64::EPSILON); + } + + #[tokio::test(flavor = "current_thread")] + async fn acp_compaction_notifies_client_of_reduced_context_and_summary_cost() { + use crate::session::types::{Message, MessagePart}; + use agent_client_protocol::Agent; + tokio::task::LocalSet::new() + .run_until(async { + let workspace = tempfile::tempdir().unwrap(); + let mut manager = SessionManager::new() + .with_history_for_workspace(workspace.path()) + .unwrap(); + let session_id = manager.create_session(Some("ACP compaction regression".into())); + let mut assistant = Message::assistant("old answer"); + assistant + .parts + .push(MessagePart::usage(80_000, 200, 0, 0, 0.25)); + let messages = vec![Message::user("history ".repeat(2_000)), assistant]; + let selection = crate::session::compaction::select_messages_for_compaction( + &messages, + crate::session::compaction::DEFAULT_TAIL_TURNS, + ) + .unwrap(); + let mut session = test_session(); + session.context_window = Some(100_000); + let summary = crate::llm::client::CompactionSummary { + text: "Retained task details".into(), + usage: crate::aisdk::chunk::TokenUsage { + input: 3_000, + output: 20, + cache_read: 0, + cache_write: 0, + }, + }; + let (mut compacted, stats) = compacted_messages( + &messages, + &selection, + &summary, + &session, + crate::session::compaction::total_context_tokens(&messages), + messages.len(), + ) + .unwrap(); + let summary_message = compacted + .iter_mut() + .find(|message| crate::session::compaction::is_compaction_summary(message)) + .unwrap(); + summary_message + .parts + .push(MessagePart::usage(0, 1, 0, 0, 0.05)); + let expected_used = + crate::session::compaction::total_context_tokens(&compacted) + 1_000; + manager + .replace_session_messages(&session_id, compacted) + .unwrap(); + let service = AcpService { + sessions: Arc::new(AsyncMutex::new(AcpSessions::default())), + session_manager: Arc::new(Mutex::new(manager)), + client_capabilities: Arc::new(Mutex::new(Default::default())), + }; + let (observed_tx, mut observed_rx) = tokio::sync::mpsc::unbounded_channel(); + let client = Client.builder().on_receive_notification( + async move |notification: SessionNotification, _connection| { + let _ = observed_tx.send(notification.update); + Ok(()) + }, + agent_client_protocol::on_receive_notification!(), + ); + Agent + .builder() + .connect_with(client, async move |connection| { + service + .notify_compaction(&session_id, &session, &connection, stats, 1_000) + .unwrap(); + let feedback = tokio::time::timeout( + std::time::Duration::from_secs(2), + observed_rx.recv(), + ) + .await + .unwrap() + .unwrap(); + assert!(matches!(feedback, SessionUpdate::AgentMessageChunk(_))); + let update = tokio::time::timeout( + std::time::Duration::from_secs(2), + observed_rx.recv(), + ) + .await + .unwrap() + .unwrap(); + let SessionUpdate::UsageUpdate(usage) = update else { + panic!("missing usage update") + }; + assert_eq!(usage.used, expected_used as u64); + assert_eq!(usage.size, 100_000); + assert!((usage.cost.unwrap().amount - 0.30).abs() < f64::EPSILON); + service + .session_manager + .lock() + .unwrap() + .try_delete_session(&session_id) + .unwrap(); + Ok(()) + }) + .await + .unwrap(); + }) + .await; + } + #[test] fn session_info_includes_hierarchy_metadata() { let now = std::time::SystemTime::now(); @@ -4301,7 +4784,7 @@ mod tests { })).unwrap(); } let stream = service.receive_prompt_stream( - &session_id, &session, &connection, &cancellation, receiver, 0, 0.0, + &session_id, &session, &connection, &cancellation, receiver, 0, 0, 0.0, ); tokio::pin!(stream); tokio::time::timeout(Duration::from_secs(2), async { diff --git a/src/session/compaction.rs b/src/session/compaction.rs index af19578..4c84a1e 100644 --- a/src/session/compaction.rs +++ b/src/session/compaction.rs @@ -546,22 +546,20 @@ fn message_parts_context_tokens(message: &Message) -> usize { } match message.role { - MessageRole::Assistant => assistant_parts_context_tokens(message), + MessageRole::Assistant => assistant_parts_context_tokens(&message.parts), MessageRole::Tool => estimate_tokens(&tool_content_for_prompt(&message.content)), _ => estimate_tokens(&message.content), } } -fn assistant_parts_context_tokens(message: &Message) -> usize { - let tool_call_ids = message - .parts +pub(crate) fn assistant_parts_context_tokens(parts: &[MessagePart]) -> usize { + let tool_call_ids = parts .iter() .filter(|part| part.part_type == "tool_call") .filter_map(|part| part.tool_id().map(|id| id.to_string())) .collect::>(); - message - .parts + parts .iter() .map(|part| match part.part_type.as_str() { "text" => part.text_value().map(estimate_tokens).unwrap_or(0), diff --git a/tests/acp_stdio.rs b/tests/acp_stdio.rs index b16988d..4e635d8 100644 --- a/tests/acp_stdio.rs +++ b/tests/acp_stdio.rs @@ -1,8 +1,160 @@ -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::{TcpListener, TcpStream}; use std::process::{Child, ChildStdin, Command, Stdio}; use std::sync::mpsc::{self, Receiver}; use std::time::{Duration, Instant}; +// The provider runs locally, while ACP itself runs in a separate process. No +// process-global environment or external API credentials are needed. +struct MockCompactionProvider { + address: std::net::SocketAddr, + requests: std::sync::Arc>>, + stop: std::sync::Arc, + thread: Option>>, +} + +impl Drop for AcpProcess { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +impl MockCompactionProvider { + fn start() -> Self { + let listener = TcpListener::bind("127.0.0.1:0").expect("mock listener"); + let address = listener.local_addr().unwrap(); + listener.set_nonblocking(true).unwrap(); + let requests = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let recorded = requests.clone(); + let stopping = stop.clone(); + let thread = std::thread::spawn(move || { + let deadline = Instant::now() + Duration::from_secs(60); + let mut regular_turns = 0; + while !stopping.load(std::sync::atomic::Ordering::Relaxed) { + if Instant::now() >= deadline { + return Err("mock provider deadline exceeded".into()); + } + match listener.accept() { + Ok((stream, _)) => { + Self::respond(stream, &recorded, &mut regular_turns) + .map_err(|error| error.to_string())?; + } + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(5)); + } + Err(error) => return Err(error.to_string()), + } + } + Ok(()) + }); + Self { + address, + requests, + stop, + thread: Some(thread), + } + } + + fn respond( + mut stream: TcpStream, + requests: &std::sync::Mutex>, + regular_turns: &mut usize, + ) -> std::io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(5)))?; + stream.set_write_timeout(Some(Duration::from_secs(5)))?; + let mut reader = BufReader::new(stream.try_clone()?); + let mut request_line = String::new(); + reader.read_line(&mut request_line)?; + let mut length = 0; + loop { + let mut header = String::new(); + if reader.read_line(&mut header)? == 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "HTTP headers", + )); + } + if header == "\r\n" { + break; + } + if let Some((name, value)) = header.split_once(':') { + if name.eq_ignore_ascii_case("content-length") { + length = value.trim().parse::().expect("HTTP content length"); + } + } + } + let (content_type, body) = if request_line.starts_with("GET /v1/models ") { + ( + "application/json", + serde_json::json!({"object": "list", "data": []}).to_string(), + ) + } else { + assert!( + request_line.starts_with("POST /v1/chat/completions "), + "unexpected request: {request_line}" + ); + assert!(length < 2_000_000, "unexpectedly large mock request"); + let mut body = vec![0; length]; + reader.read_exact(&mut body)?; + let request: serde_json::Value = serde_json::from_slice(&body).expect("provider JSON"); + let compacting = request["messages"] + .to_string() + .contains("CONTEXT CHECKPOINT COMPACTION"); + requests.lock().unwrap().push(request); + let (text, input) = if compacting { + ( + "Checkpoint: the user requested a brief answer. Continue with the next task.", + 300, + ) + } else { + *regular_turns += 1; + ( + "Mock answer.", + if *regular_turns == 1 { 85_000 } else { 500 }, + ) + }; + let chunk = serde_json::json!({ + "id": "mock-completion", "object": "chat.completion.chunk", + "created": 0, "model": "chat", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": text}, "finish_reason": null}] + }); + let end = serde_json::json!({ + "id": "mock-completion", "object": "chat.completion.chunk", + "created": 0, "model": "chat", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": input, "completion_tokens": 100, "total_tokens": input + 100} + }); + ( + "text/event-stream", + format!("data: {chunk}\n\ndata: {end}\n\ndata: [DONE]\n\n"), + ) + }; + write!(stream, "HTTP/1.1 200 OK\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len())?; + stream.flush() + } + + fn finish(mut self) { + self.stop.store(true, std::sync::atomic::Ordering::Relaxed); + self.thread + .take() + .unwrap() + .join() + .expect("mock provider thread") + .expect("mock provider HTTP handling"); + } +} + +impl Drop for MockCompactionProvider { + fn drop(&mut self) { + self.stop.store(true, std::sync::atomic::Ordering::Relaxed); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + struct AcpProcess { child: Child, stdin: Option, @@ -14,6 +166,13 @@ struct AcpProcess { impl AcpProcess { fn spawn(workspace: &std::path::Path) -> Self { + Self::spawn_with_working_directory(workspace, &std::env::current_dir().unwrap()) + } + + fn spawn_with_working_directory( + workspace: &std::path::Path, + working_directory: &std::path::Path, + ) -> Self { let home = tempfile::tempdir().expect("home"); let config = tempfile::tempdir().expect("config"); let state = tempfile::tempdir().expect("state"); @@ -33,6 +192,7 @@ impl AcpProcess { let mut child = Command::new(env!("CARGO_BIN_EXE_crabcode")) .args(["acp", "--cwd"]) .arg(workspace) + .current_dir(working_directory) .env("HOME", home.path()) .env("XDG_CONFIG_HOME", config.path()) .env("XDG_STATE_HOME", state.path()) @@ -320,6 +480,23 @@ fn session_load_error_keeps_stdio_transport_open() { #[test] fn advertises_and_dispatches_commands_over_stdio() { let workspace = tempfile::tempdir().expect("workspace"); + // The empty models.dev cache avoids external catalog requests, but session + // setup still needs a configured model. These commands never call a provider; + // omitting baseURL also avoids custom-provider model discovery requests. + std::fs::write( + workspace.path().join("crabcode.json"), + serde_json::json!({ + "model": "offline/chat", + "enabled_providers": ["offline"], + "provider": {"offline": { + "npm": "@ai-sdk/openai-compatible", + "options": {"apiKey": "dummy"}, + "models": {"chat": {"limit": {"context": 100000, "output": 1000}}} + }} + }) + .to_string(), + ) + .expect("offline workspace config"); let command_dir = workspace.path().join(".crabcode/commands"); std::fs::create_dir_all(&command_dir).expect("command dir"); std::fs::write( @@ -441,3 +618,131 @@ Wire skill instructions. process.close_and_wait(); } + +#[test] +fn automatically_compacts_provider_usage_over_stdio() { + let workspace = tempfile::tempdir().expect("workspace"); + let provider = MockCompactionProvider::start(); + std::fs::write( + workspace.path().join("crabcode.json"), + serde_json::json!({ + "model": "mock/chat", + "enabled_providers": ["mock"], + "provider": {"mock": { + "npm": "@ai-sdk/openai-compatible", + "options": {"baseURL": format!("http://{}/v1", provider.address), "apiKey": "dummy"}, + "models": {"chat": {"limit": {"context": 100000, "output": 1000}}} + }}, + "compaction": {"auto": true, "reserved": 20000} + }).to_string(), + ).expect("workspace config"); + let system_marker = "ACP_MOCK_WORKSPACE_INSTRUCTIONS_PRESERVED"; + std::fs::write(workspace.path().join("AGENTS.md"), system_marker).unwrap(); + // Request-time provider discovery also reads the process working directory. + let mut process = AcpProcess::spawn_with_working_directory(workspace.path(), workspace.path()); + assert!(initialize(&mut process).get("error").is_none()); + process.send(serde_json::json!({ + "jsonrpc": "2.0", "id": 2, "method": "session/new", + "params": {"cwd": workspace.path(), "mcpServers": []} + })); + let (response, _) = process.recv_response(2); + let session_id = response["result"]["sessionId"] + .as_str() + .expect("session id"); + // 3,000 estimated tokens clears the 2,000-token compaction minimum but + // cannot reach the 80,000-token threshold without authoritative usage. + let initial_prompt = format!( + "ARCHIVED_LARGE_INITIAL_PROMPT {}Please answer briefly.", + "word ".repeat(2400) + ); + assert!(initial_prompt.len() / 4 >= 2000); + assert!(initial_prompt.len() / 4 < 80000); + process.send(serde_json::json!({ + "jsonrpc": "2.0", "id": 3, "method": "session/prompt", + "params": {"sessionId": session_id, "prompt": [{"type": "text", "text": initial_prompt}]} + })); + let (response, notifications) = process.recv_response(3); + assert_eq!(response["result"]["stopReason"], "end_turn", "{response}"); + let compacted = notifications + .iter() + .position(|message| { + message["params"]["update"]["content"]["text"] + .as_str() + .is_some_and(|text| text.contains("Context compacted")) + }) + .expect("automatic compaction notification without /compact"); + let usage = notifications + .iter() + .filter(|message| message["params"]["update"]["sessionUpdate"] == "usage_update") + .collect::>(); + assert!( + usage + .iter() + .any(|message| message["params"]["update"]["used"] + .as_u64() + .is_some_and(|used| used >= 85000)), + "provider usage must reach ACP: {usage:?}" + ); + let final_usage = &usage.last().expect("final usage")["params"]["update"]; + assert_eq!(final_usage["size"], 100000); + assert!( + final_usage["used"].as_u64().unwrap() < 80000, + "{final_usage}" + ); + assert!(notifications[compacted + 1..] + .iter() + .any(|message| message["params"]["update"]["sessionUpdate"] == "usage_update")); + { + let requests = provider.requests.lock().unwrap(); + assert_eq!( + requests.len(), + 2, + "one regular request then one summary request" + ); + assert!( + requests[0]["messages"].to_string().chars().count() / 4 < 80_000, + "estimated initial transcript alone must not trigger compaction" + ); + assert!(!requests[0]["messages"] + .to_string() + .contains("CONTEXT CHECKPOINT COMPACTION")); + assert!(requests[1]["messages"] + .to_string() + .contains("CONTEXT CHECKPOINT COMPACTION")); + } + process.send(serde_json::json!({ + "jsonrpc": "2.0", "id": 4, "method": "session/prompt", + "params": {"sessionId": session_id, "prompt": [{"type": "text", "text": "Now give another brief answer."}]} + })); + let (response, notifications) = process.recv_response(4); + assert_eq!(response["result"]["stopReason"], "end_turn", "{response}"); + assert!( + !notifications + .iter() + .any(|message| message["params"]["update"]["content"]["text"] + .as_str() + .is_some_and(|text| text.contains("Context compacted"))), + "stale usage must not compact the second turn" + ); + { + let requests = provider.requests.lock().unwrap(); + assert_eq!(requests.len(), 3, "no repeated compaction from stale usage"); + let messages = requests[2]["messages"] + .as_array() + .expect("second-turn messages"); + assert_eq!(messages[0]["role"], "system"); + assert!( + messages[0]["content"].to_string().contains(system_marker), + "fresh system prompt preserves workspace instructions" + ); + let text = requests[2]["messages"].to_string(); + assert!(text.contains("Checkpoint:"), "summary is active context"); + assert!( + !text.contains("ARCHIVED_LARGE_INITIAL_PROMPT"), + "archived transcript must not be resent" + ); + assert!(!text.contains("CONTEXT CHECKPOINT COMPACTION")); + } + process.close_and_wait(); + provider.finish(); +} From 4a4f25a415d14a8e6cdeded0bf482a7abcad7b2c Mon Sep 17 00:00:00 2001 From: Yanuar Date: Tue, 13 Oct 2026 11:44:43 +0700 Subject: [PATCH 2/2] fix(acp): describe configured compaction mode in picker --- src/acp/service.rs | 84 ++++++++++++++++++++++++++++++++++++++++++++-- tests/acp_stdio.rs | 23 +++++++++++++ 2 files changed, 105 insertions(+), 2 deletions(-) diff --git a/src/acp/service.rs b/src/acp/service.rs index 97f3380..e0fa41d 100644 --- a/src/acp/service.rs +++ b/src/acp/service.rs @@ -2262,6 +2262,14 @@ fn tool_permissions(session: &AcpSession) -> crate::tools::ToolPermissions { session.tool_permissions.clone() } +fn mode_description(agent: &crate::agent::definition::AgentDefinition) -> &str { + if agent.name == "compaction" && agent.description.trim().is_empty() { + "Compaction agent. Summarizes conversation context for continued work." + } else { + &agent.description + } +} + fn session_modes(session: &AcpSession) -> SessionModeState { let mut modes = session .config @@ -2271,7 +2279,7 @@ fn session_modes(session: &AcpSession) -> SessionModeState { .into_iter() .map(|agent| { SessionMode::new(agent.name.clone(), agent.name.clone()) - .description(agent.description.clone()) + .description(mode_description(agent)) }) .collect::>(); modes.sort_by(|left, right| left.name.cmp(&right.name)); @@ -2287,7 +2295,7 @@ fn session_config_options(session: &AcpSession) -> Vec { .into_iter() .map(|agent| { SessionConfigSelectOption::new(agent.name.clone(), agent.name.clone()) - .description(agent.description.clone()) + .description(mode_description(agent)) }) .collect::>(); mode_options.sort_by(|left, right| left.name.cmp(&right.name)); @@ -5114,6 +5122,78 @@ mod tests { assert!(workspace_path(Path::new("relative")).is_err()); } + #[test] + fn compaction_mode_descriptions_preserve_configuration_and_visibility() { + let mut session = test_session(); + assert!(!session_modes(&session) + .available_modes + .iter() + .any(|mode| mode.id.to_string() == "compaction")); + + for description in ["", " \n ", "Use this agent to preserve project decisions."] { + let definitions = crate::agent::definition::parse_agent_definitions_from_config( + Some(&serde_json::json!({"compaction": { + "mode": "primary", "description": description + }})), + &mut Vec::new(), + ); + session.config.merged_config.agent_registry = + crate::agent::definition::AgentRegistry::with_definitions(None, definitions); + let expected = if description.trim().is_empty() { + "Compaction agent. Summarizes conversation context for continued work." + } else { + description + }; + let modes = session_modes(&session); + let compaction = modes + .available_modes + .iter() + .find(|mode| mode.id.to_string() == "compaction") + .expect("configured compaction mode"); + assert_eq!(compaction.description.as_deref(), Some(expected)); + let config = serde_json::to_value(session_config_options(&session)).unwrap(); + let compaction = config[0]["options"] + .as_array() + .unwrap() + .iter() + .find(|option| option["value"] == "compaction") + .expect("configured compaction option"); + assert_eq!(compaction["description"], expected); + assert_eq!( + modes + .available_modes + .iter() + .find(|mode| mode.id.to_string() == "build") + .unwrap() + .description + .as_deref(), + Some("The default agent. Executes tools based on configured permissions.") + ); + } + + for settings in [ + serde_json::json!({"mode": "subagent"}), + serde_json::json!({"mode": "primary", "hidden": true}), + ] { + let definitions = crate::agent::definition::parse_agent_definitions_from_config( + Some(&serde_json::json!({"compaction": settings})), + &mut Vec::new(), + ); + session.config.merged_config.agent_registry = + crate::agent::definition::AgentRegistry::with_definitions(None, definitions); + assert!(!session_modes(&session) + .available_modes + .iter() + .any(|mode| mode.id.to_string() == "compaction")); + let config = serde_json::to_value(session_config_options(&session)).unwrap(); + assert!(!config[0]["options"] + .as_array() + .unwrap() + .iter() + .any(|option| option["value"] == "compaction")); + } + } + #[test] fn builds_grouped_model_config_option() { let option = model_config_option( diff --git a/tests/acp_stdio.rs b/tests/acp_stdio.rs index 4e635d8..26a17a2 100644 --- a/tests/acp_stdio.rs +++ b/tests/acp_stdio.rs @@ -488,6 +488,7 @@ fn advertises_and_dispatches_commands_over_stdio() { serde_json::json!({ "model": "offline/chat", "enabled_providers": ["offline"], + "agent": {"compaction": {"mode": "primary"}}, "provider": {"offline": { "npm": "@ai-sdk/openai-compatible", "options": {"apiKey": "dummy"}, @@ -534,6 +535,28 @@ Wire skill instructions. } })); let (new_session, mut notifications) = process.recv_response(2); + let expected_description = + "Compaction agent. Summarizes conversation context for continued work."; + let compaction_mode = new_session["result"]["modes"]["availableModes"] + .as_array() + .expect("available modes") + .iter() + .find(|mode| mode["id"] == "compaction") + .expect("compaction mode"); + assert_eq!(compaction_mode["description"], expected_description); + let mode_config = new_session["result"]["configOptions"] + .as_array() + .expect("config options") + .iter() + .find(|option| option["id"] == "mode") + .expect("mode config"); + let compaction_option = mode_config["options"] + .as_array() + .expect("mode options") + .iter() + .find(|option| option["value"] == "compaction") + .expect("compaction option"); + assert_eq!(compaction_option["description"], expected_description); let session_id = new_session["result"]["sessionId"] .as_str() .expect("session id")