From cec35bfbb19109f8d43f7b5d5394b45d8ef326cb Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 30 Sep 2026 00:41:05 +0200 Subject: [PATCH] hive-runtime: compact ACP sessions through the agent's compact command An ACP agent's session is now compacted like a claude one: proactively once a turn crosses the percent-of-window watermark, and on the operator's /compact or the agent's compact tool. Before, the ACP backend's compact returned Unsupported and no watermark applied to it. - AcpRuntime takes the same CompactionPolicy as ClaudeRuntime; make_session builds one PercentPolicy (with CHECKPOINT_PROMPT) and hands it to whichever backend runs. - The runtime keeps the commands each session advertises in available_commands_update. If `compact` is among them, compaction sends the prompt `/compact` on the same session, which is how the ACP spec runs an advertised command. A proactive compaction runs the checkpoint turn first, as InfiniteSession does. - With no compact command, the checkpoint turn runs, the session is archived, and the next turn starts a new one with the system prompt. - Error::Unsupported had no producer left, so it and drive_turn's "/compact skipped" arm are gone. Refs #4391 --- docs/turn-loop/claude-invocation.md | 3 +- hive-agent/src/turn.rs | 45 ++-- hive-runtime/README.md | 13 +- hive-runtime/src/acp/mod.rs | 347 ++++++++++++++++++++++++---- hive-runtime/src/acp/stream.rs | 17 ++ hive-runtime/src/lib.rs | 9 +- 6 files changed, 357 insertions(+), 77 deletions(-) diff --git a/docs/turn-loop/claude-invocation.md b/docs/turn-loop/claude-invocation.md index bfb5f4e1..edd6d0d1 100644 --- a/docs/turn-loop/claude-invocation.md +++ b/docs/turn-loop/claude-invocation.md @@ -161,7 +161,8 @@ model-reported context fill reaches `HIVE_COMPACT_WATERMARK_PERCENT` for the window on turns the model didn't report one. `0` disables proactive compaction (the reactive path always applies). The proactive path is best-effort — a failed checkpoint or `/compact` never fails the turn that -already succeeded. +already succeeded. An ACP agent gets the same policy; how it compacts is in +[`hive-runtime/README.md`](../../hive-runtime/README.md#acp-backend-compaction). The operator can force a compaction any time via `POST /api/compact`. It's **deferred**: the handler sets `Bus::request_compact()` and returns diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index 6b007318..ee970c70 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -305,15 +305,16 @@ fn compact_percent() -> u8 { const ACP_SESSION_FILE: &str = "hyperhive-acp-session"; /// The agent's durable session, on the runtime its environment selects -/// (`hive_runtime::RuntimeSpec`). On claude it is the constant-title -/// [`hive_claude::InfiniteSession`] with hyperhive's percent-of-window -/// compaction policy. Built once by the serve loop (see [`make_session`]) and -/// threaded through the turns, rather than rebuilt each time. +/// (`hive_runtime::RuntimeSpec`), with hyperhive's percent-of-window +/// compaction policy. On claude it is the constant-title +/// [`hive_claude::InfiniteSession`]. Built once by the serve loop (see +/// [`make_session`]) and threaded through the turns, rather than rebuilt each +/// time. pub type AgentSession = AgentRuntime; -/// Construct the agent's durable session. On claude: constant title + on-disk -/// store + a percent-of-window compaction policy that checkpoints -/// (`CHECKPOINT_PROMPT`) before compacting. Called once at serve-loop start. +/// Construct the agent's durable session, with a percent-of-window compaction +/// policy that checkpoints (`CHECKPOINT_PROMPT`) before compacting; on claude +/// also a constant title and the on-disk store. Called once at serve-loop start. /// `percent` comes from a boot-time env var and `default_window` is only a /// fallback for turns where the model didn't report a window, so a single /// build at startup is fine. @@ -322,20 +323,20 @@ pub type AgentSession = AgentRuntime; /// /// Returns an error if the runtime selection in the environment is invalid. pub fn make_session(bus: &Bus) -> Result { + let policy = PercentPolicy { + percent: compact_percent(), + default_window: Some(effective_context_window(bus)), + checkpoint_prompt: Some(CHECKPOINT_PROMPT.to_string()), + }; Ok(match RuntimeSpec::from_env()? { - RuntimeSpec::Claude => AgentRuntime::Claude(ClaudeRuntime::new( - session_title(), - session_store(), - PercentPolicy { - percent: compact_percent(), - default_window: Some(effective_context_window(bus)), - checkpoint_prompt: Some(CHECKPOINT_PROMPT.to_string()), - }, - )), + RuntimeSpec::Claude => { + AgentRuntime::Claude(ClaudeRuntime::new(session_title(), session_store(), policy)) + } RuntimeSpec::Acp(command) => AgentRuntime::Acp(Box::new(AcpRuntime::new( command, crate::paths::harness_dir().join(ACP_SESSION_FILE), acp_permission_policy(), + policy, ))), }) } @@ -460,15 +461,7 @@ pub async fn drive_turn( // Reflect `Compacting` in the UI like the idle path (`run_pending_compact`) // does; the serve loop resets to `Idle` once this turn returns. bus.set_state(crate::events::TurnState::Compacting); - let compacted = match session.compact(&config, &sink).await { - Err(e @ hive_runtime::Error::Unsupported(_)) => { - bus.emit(LiveEvent::Note { - text: format!("/compact skipped: {e}"), - }); - false - } - _ => true, - }; + let _ = session.compact(&config, &sink).await; // If the compact call asked to be woken (the agent's own `compact` // tool with a `wake_prompt`), stash it — the serve loop reads it // back after this turn returns and drives a synthetic follow-up @@ -477,7 +470,7 @@ pub async fn drive_turn( if let Some(prompt) = request.wake_prompt { bus.set_post_compact_wake(prompt); } - return Ok(compacted); + return Ok(true); } outcome } diff --git a/hive-runtime/README.md b/hive-runtime/README.md index 581146a7..4ba35db0 100644 --- a/hive-runtime/README.md +++ b/hive-runtime/README.md @@ -40,6 +40,15 @@ with `AcpError::IdleTimeout` (`IdleKilled` if the agent was killed). This is the only way a provider error the agent retries without reporting it, such as an HTTP 429, ends the turn. -## ACP backend: not yet +## ACP backend: compaction -`compact` returns `Error::Unsupported`. +The same `CompactionPolicy` as the claude backend decides when: after a turn +past the watermark, or on `compact` (the operator's `/compact`, the agent's +`compact` tool). + +- If the agent advertises a `compact` command (`available_commands_update`), + it is sent as the prompt `/compact` on the same session, as the ACP spec + runs any advertised command. A proactive compaction runs the policy's + checkpoint turn first, as on claude. +- Otherwise the policy's checkpoint turn runs, then the session is archived, + and the next turn starts a new one, carrying the system prompt again. diff --git a/hive-runtime/src/acp/mod.rs b/hive-runtime/src/acp/mod.rs index 7ebf3505..c1e4e5d5 100644 --- a/hive-runtime/src/acp/mod.rs +++ b/hive-runtime/src/acp/mod.rs @@ -11,7 +11,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; -use hive_claude::{Config, Progress, Sink}; +use hive_claude::{CompactionPolicy, Config, Progress, Sink}; use serde_json::{Value, json}; use tokio::sync::futures::Notified; use tokio::sync::{Mutex, Notify}; @@ -37,6 +37,14 @@ const SETTLE_MAX: Duration = Duration::from_secs(5); /// killed. const CANCEL_GRACE: Duration = Duration::from_secs(10); +/// The command an agent advertises to compact its session. +const COMPACT_COMMAND: &str = "compact"; + +/// How long to wait, on a session whose agent has not advertised its +/// commands yet, for it to do so. An agent may send them only after the +/// session is attached, or never. +const COMMANDS_WAIT: Duration = SETTLE_MAX; + /// One `session/request_permission` request, as a [`PermissionPolicy`] sees it. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct PermissionAsk<'a> { @@ -155,10 +163,17 @@ enum Stop { /// `session/update` for that long the turn is cancelled, and fails with /// [`AcpError::IdleTimeout`]). Everything else in it is claude's and is /// ignored. -pub struct AcpRuntime { +/// +/// Compaction, proactive once `policy` says so after a turn or on +/// [`Runtime::compact`], runs the agent's advertised `compact` command as a +/// prompt on the same session. An agent advertising none gets the policy's +/// checkpoint turn instead, and the session is archived, so the next turn +/// starts a new one. +pub struct AcpRuntime { command: AcpCommand, session_file: PathBuf, permit: PermissionPolicy, + policy: P, live: Mutex>, cancel: Arc, cancel_grace: Duration, @@ -171,17 +186,27 @@ struct Live { /// The recorded session, or a new one whose first prompt is unanswered. loaded: Option, model: Option, + /// The commands the agent last advertised for `loaded`; `None` until it + /// has. + commands: Option>, } -impl AcpRuntime { +impl AcpRuntime

{ /// A runtime spawning `command`, keeping its session id in - /// `session_file`, and answering its permission requests with `permit`. + /// `session_file`, answering its permission requests with `permit`, and + /// compacting per `policy`. #[must_use] - pub fn new(command: AcpCommand, session_file: PathBuf, permit: PermissionPolicy) -> Self { + pub fn new( + command: AcpCommand, + session_file: PathBuf, + permit: PermissionPolicy, + policy: P, + ) -> Self { Self { command, session_file, permit, + policy, live: Mutex::new(None), cancel: Arc::default(), cancel_grace: CANCEL_GRACE, @@ -223,6 +248,7 @@ impl AcpRuntime { load_session: caps["loadSession"] == Value::Bool(true), loaded: None, model: None, + commands: None, }) } @@ -253,9 +279,10 @@ impl AcpRuntime { let params = json!({ "sessionId": id, "cwd": cwd, "mcpServers": servers }); match live.conn.request("session/load", params).await { Ok(response) => { - discard_stale(&mut live.conn)?; live.model = stream::session_model(&response); live.loaded = Some(id.clone()); + live.commands = None; + discard_stale(live)?; return Ok((id, new)); } Err(e @ AcpError::Rpc { .. }) => { @@ -275,6 +302,7 @@ impl AcpRuntime { .to_owned(); live.model = stream::session_model(&response); live.loaded = Some(id.clone()); + live.commands = None; write_id(&self.pending_file(), &id)?; Ok((id, true)) } @@ -297,7 +325,7 @@ impl AcpRuntime { let (session, created) = self.attach(live, &cwd, &servers).await?; let text = prompt_text(config, prompt, created); let params = json!({ "sessionId": session, "prompt": [{ "type": "text", "text": text }] }); - discard_stale(&mut live.conn)?; + discard_stale(live)?; let mut reply = live.conn.send("session/prompt", params).await?; let mut mapper = StreamMapper::new(names); let mut stderr_tail = VecDeque::new(); @@ -315,7 +343,7 @@ impl AcpRuntime { if matches!(incoming, Incoming::Update(_)) { quiet_until = idle.map(|window| Instant::now() + window); } - if !deliver(incoming, &session, &mut mapper, &mut stderr_tail, sink) { + if !deliver(incoming, &mut mapper, live, &mut stderr_tail, sink) { let stderr_tail = Vec::from(stderr_tail).join("\n"); return Err(AcpError::Exited { stderr_tail }.into()); } @@ -360,7 +388,7 @@ impl AcpRuntime { ) .await { - if !deliver(incoming, &session, &mut mapper, &mut stderr_tail, sink) { + if !deliver(incoming, &mut mapper, live, &mut stderr_tail, sink) { break; } } @@ -391,17 +419,33 @@ impl AcpRuntime { progress.telemetry = mapper.telemetry(&response, live.model.as_deref()); Ok(progress) } -} -impl Runtime for AcpRuntime { - async fn run(&self, config: &Config, prompt: &str, sink: &impl Sink) -> Result { - let mut guard = self.live.lock().await; - if let Some(live) = guard.as_mut() - && live.conn.exited() - { + /// The running agent, spawned if there is none or it has exited. + async fn running<'g>( + &self, + guard: &'g mut Option, + config: &Config, + ) -> Result<&'g mut Live> { + let mut live = guard.take(); + if live.as_mut().is_some_and(|live| live.conn.exited()) { tracing::warn!("ACP agent has exited; respawning"); - *guard = None; + live = None; } + let live = match live { + Some(live) => live, + None => self.start(config).await?, + }; + Ok(guard.insert(live)) + } + + /// Run `prompt` as one turn on the durable session. + async fn prompt( + &self, + guard: &mut Option, + config: &Config, + prompt: &str, + sink: &impl Sink, + ) -> Result { // Registered before the turn is marked in flight, so a cancel is // never missed; one that comes before the prompt is sent stops it // right after. @@ -410,10 +454,7 @@ impl Runtime for AcpRuntime { cancelled.as_mut().enable(); self.cancel.in_turn.store(true, Ordering::SeqCst); let _in_turn = InTurn(&self.cancel.in_turn); - let live = match guard.as_mut() { - Some(live) => live, - None => guard.insert(self.start(config).await?), - }; + let live = self.running(guard, config).await?; let result = self.turn(live, config, prompt, sink, cancelled).await; if let Err(Error::Acp( AcpError::Closed @@ -429,8 +470,65 @@ impl Runtime for AcpRuntime { result } - async fn compact(&self, _config: &Config, _sink: &impl Sink) -> Result<()> { - Err(Error::Unsupported("compact")) + /// Compact the recorded session, if there is one. `checkpoint` runs the + /// policy's checkpoint turn before an advertised `compact` command too; + /// without one it always runs, since archiving keeps nothing of the + /// session. + async fn compact_session( + &self, + guard: &mut Option, + config: &Config, + sink: &impl Sink, + checkpoint: bool, + ) -> Result<()> { + if read_id(&self.session_file).is_none() { + return Ok(()); + } + let live = self.running(guard, config).await?; + let (servers, _) = mcp_servers(config)?; + self.attach(live, &session_cwd(config), &servers).await?; + if advertises_compact(live).await? { + if checkpoint { + self.checkpoint(guard, config, sink).await; + } + self.prompt(guard, config, &format!("/{COMPACT_COMMAND}"), sink) + .await?; + return Ok(()); + } + tracing::info!("ACP agent advertises no compact command; starting a new session"); + self.checkpoint(guard, config, sink).await; + self.archive()?; + Ok(()) + } + + /// The policy's checkpoint turn, if it has one. Best-effort: compaction + /// goes ahead if it fails. + async fn checkpoint(&self, guard: &mut Option, config: &Config, sink: &impl Sink) { + let Some(prompt) = self.policy.checkpoint_prompt() else { + return; + }; + if let Err(e) = self.prompt(guard, config, prompt, sink).await { + tracing::warn!(error = %e, "ACP checkpoint turn failed"); + } + } +} + +impl Runtime for AcpRuntime

{ + async fn run(&self, config: &Config, prompt: &str, sink: &impl Sink) -> Result { + let mut guard = self.live.lock().await; + let mut progress = self.prompt(&mut guard, config, prompt, sink).await?; + if self.policy.should_compact(progress.telemetry.usage()) { + match self.compact_session(&mut guard, config, sink, true).await { + Ok(()) => progress.compacted = true, + Err(e) => tracing::warn!(error = %e, "ACP compaction failed"), + } + } + Ok(progress) + } + + async fn compact(&self, config: &Config, sink: &impl Sink) -> Result<()> { + let mut guard = self.live.lock().await; + self.compact_session(&mut guard, config, sink, false).await } fn canceller(&self) -> Option { @@ -456,18 +554,22 @@ impl Runtime for AcpRuntime { } } -/// Hand one incoming item to the sink. Returns `false` once the agent has +/// Hand one incoming item to the sink, and keep the commands the agent +/// advertises for the loaded session. Returns `false` once the agent has /// closed its output. fn deliver( incoming: Incoming, - session: &str, mapper: &mut StreamMapper, + agent: &mut Live, stderr_tail: &mut VecDeque, sink: &impl Sink, ) -> bool { match incoming { Incoming::Update(params) => { - if params["sessionId"].as_str() == Some(session) { + if params["sessionId"].as_str() == agent.loaded.as_deref() { + if let Some(advertised) = stream::advertised_commands(¶ms["update"]) { + agent.commands = Some(advertised); + } for event in mapper.push(¶ms["update"]) { sink.on_event(&event); } @@ -489,19 +591,49 @@ fn deliver( /// Drop what the agent sent outside a turn: the history a `session/load` /// replays (the caller already has it), and anything that arrived after the /// previous turn settled, which must not be shown as part of the next one. -fn discard_stale(conn: &mut Connection) -> std::result::Result<(), AcpError> { - while let Ok(incoming) = conn.incoming.try_recv() { - match incoming { - Incoming::Update(_) => {} - Incoming::Stdout(line) | Incoming::Stderr(line) => { - tracing::info!(line = %line, "ACP agent output between turns"); - } - Incoming::Closed => return Err(AcpError::Closed), - } +/// The commands the agent advertises for the loaded session are kept. +fn discard_stale(live: &mut Live) -> std::result::Result<(), AcpError> { + while let Ok(incoming) = live.conn.incoming.try_recv() { + between_turns(live, incoming)?; } Ok(()) } +fn between_turns(agent: &mut Live, incoming: Incoming) -> std::result::Result<(), AcpError> { + match incoming { + Incoming::Update(params) => { + if params["sessionId"].as_str() == agent.loaded.as_deref() + && let Some(advertised) = stream::advertised_commands(¶ms["update"]) + { + agent.commands = Some(advertised); + } + } + Incoming::Stdout(line) | Incoming::Stderr(line) => { + tracing::info!(line = %line, "ACP agent output between turns"); + } + Incoming::Closed => return Err(AcpError::Closed), + } + Ok(()) +} + +/// Whether the agent advertises a `compact` command for the loaded session, +/// waiting up to [`COMMANDS_WAIT`] for its commands if it has sent none yet. +async fn advertises_compact(live: &mut Live) -> std::result::Result { + discard_stale(live)?; + let deadline = Instant::now() + COMMANDS_WAIT; + while live.commands.is_none() { + match tokio::time::timeout_at(deadline, live.conn.incoming.recv()).await { + Ok(incoming) => between_turns(live, incoming.unwrap_or(Incoming::Closed))?, + Err(_) => break, + } + } + Ok(live + .commands + .iter() + .flatten() + .any(|name| name == COMPACT_COMMAND)) +} + /// The prompt as sent: the first one of a session carries the system prompt. fn prompt_text(config: &Config, prompt: &str, first: bool) -> String { match (first, &config.system_prompt_file) { @@ -566,7 +698,7 @@ mod tests { use std::sync::Arc; use std::time::Duration; - use hive_claude::{Config, NoopSink, Sink}; + use hive_claude::{Config, NoopSink, PercentPolicy, Sink}; use super::{AcpError, AcpRuntime}; use crate::{AcpCommand, Error, Runtime}; @@ -583,24 +715,39 @@ mod tests { /// reply an agent gives a cancelled prompt it was retrying against a /// rate-limited (HTTP 429) provider without reporting it; /// - `deaf`: nothing, and `session/cancel` is ignored; - /// - `trickle`: five text chunks 100ms apart, then `end_turn`. + /// - `trickle`: five text chunks 100ms apart, then `end_turn`; + /// - `ok`: `end_turn`, like the rest. + /// + /// With `COMMANDS` set, it advertises that one command on each session it + /// creates or loads. With `USED` set, it reports that many of 1000 context + /// tokens used on each prompt. const AGENT: &str = r#" n=0 prompt= printf 'start\n' >> "$1.methods" +update() { + printf '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"%s","update":%s}}\n' "$1" "$2" +} +advertise() { + [ -n "$COMMANDS" ] && update "$1" "{\"sessionUpdate\":\"available_commands_update\",\"availableCommands\":[{\"name\":\"$COMMANDS\",\"description\":\"\"}]}" +} while IFS= read -r line; do id=${line#*\"id\":}; id=${id%%[,\}]*} m=${line#*\"method\":\"}; m=${m%%\"*} + sid=${line#*\"sessionId\":\"}; sid=${sid%%\"*} printf '%s\n' "$m" >> "$1.methods" case $m in initialize) printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":1,"agentCapabilities":{"loadSession":true,"mcpCapabilities":{"http":true}}}}\n' "$id" ;; session/new) n=$((n+1)) - printf '{"jsonrpc":"2.0","id":%s,"result":{"sessionId":"s%s"}}\n' "$id" "$n" ;; + printf '{"jsonrpc":"2.0","id":%s,"result":{"sessionId":"s%s"}}\n' "$id" "$n" + advertise "s$n" ;; session/load) - printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;; + printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" + advertise "$sid" ;; session/prompt) printf '%s\n' "$line" >> "$1" + [ -n "$USED" ] && update "$sid" "{\"sessionUpdate\":\"usage_update\",\"used\":$USED,\"size\":1000}" if [ -e "$1.once" ]; then printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id" else @@ -608,6 +755,8 @@ while IFS= read -r line; do case $2 in fail) printf '{"jsonrpc":"2.0","id":%s,"error":{"code":-32603,"message":"provider down"}}\n' "$id" ;; + ok) + printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id" ;; silent) prompt=$id ;; trickle) for _ in 1 2 3 4 5; do @@ -626,7 +775,16 @@ while IFS= read -r line; do done "#; - fn runtime(dir: &Path, mode: &str) -> AcpRuntime { + fn runtime(dir: &Path, mode: &str) -> AcpRuntime { + agent(dir, mode, &[], PercentPolicy::default()) + } + + fn agent( + dir: &Path, + mode: &str, + env: &[(&str, &str)], + policy: PercentPolicy, + ) -> AcpRuntime { let script = dir.join("agent.sh"); std::fs::write(&script, AGENT).unwrap(); let command = AcpCommand { @@ -636,9 +794,43 @@ done dir.join("prompts").display().to_string(), mode.to_owned(), ], - env: BTreeMap::new(), + env: env + .iter() + .map(|(k, v)| ((*k).to_owned(), (*v).to_owned())) + .collect::>(), }; - AcpRuntime::new(command, dir.join("session"), Arc::new(|_: &_| false)) + AcpRuntime::new( + command, + dir.join("session"), + Arc::new(|_: &_| false), + policy, + ) + } + + /// Compacts at `percent` of the window, after a `CHECKPOINT` turn. + fn policy(percent: u8) -> PercentPolicy { + PercentPolicy { + percent, + default_window: None, + checkpoint_prompt: Some("CHECKPOINT".to_owned()), + } + } + + /// Each prompt the agent was sent: its session and its text. + fn prompts(dir: &Path) -> Vec<(String, String)> { + std::fs::read_to_string(dir.join("prompts")) + .unwrap() + .lines() + .map(|line| { + let request: serde_json::Value = serde_json::from_str(line).unwrap(); + let params = &request["params"]; + let text = params["prompt"][0]["text"].as_str().unwrap(); + ( + params["sessionId"].as_str().unwrap().to_owned(), + text.to_owned(), + ) + }) + .collect() } fn config(dir: &Path) -> Config { @@ -838,4 +1030,75 @@ done runtime.run(&config, "two", &NoopSink).await.unwrap(); assert_eq!(count(dir.path(), "start"), 2); } + + fn sent(session: &str, text: &str) -> (String, String) { + (session.to_owned(), text.to_owned()) + } + + #[tokio::test] + async fn an_advertised_compact_command_runs_on_the_same_session() { + let dir = tempfile::tempdir().unwrap(); + let runtime = agent(dir.path(), "ok", &[("COMMANDS", "compact")], policy(0)); + let config = config(dir.path()); + + runtime.run(&config, "one", &NoopSink).await.unwrap(); + runtime.compact(&config, &NoopSink).await.unwrap(); + let next = runtime.run(&config, "two", &NoopSink).await.unwrap(); + + assert!(!next.created); + assert_eq!( + prompts(dir.path()), + [ + sent("s1", "SYSTEM PROMPT\n\none"), + sent("s1", "/compact"), + sent("s1", "two"), + ] + ); + assert_eq!(count(dir.path(), "session/new"), 1); + assert_eq!(recorded(dir.path()), "s1"); + } + + #[tokio::test] + async fn with_no_compact_command_notes_are_written_then_a_new_session_starts() { + let dir = tempfile::tempdir().unwrap(); + let runtime = agent(dir.path(), "ok", &[("COMMANDS", "init")], policy(0)); + let config = config(dir.path()); + + runtime.run(&config, "one", &NoopSink).await.unwrap(); + runtime.compact(&config, &NoopSink).await.unwrap(); + let next = runtime.run(&config, "two", &NoopSink).await.unwrap(); + + assert!(next.created); + assert_eq!( + prompts(dir.path()), + [ + sent("s1", "SYSTEM PROMPT\n\none"), + sent("s1", "CHECKPOINT"), + sent("s2", "SYSTEM PROMPT\n\ntwo"), + ] + ); + assert_eq!(recorded(dir.path()), "s2"); + } + + #[tokio::test] + async fn a_turn_past_the_watermark_checkpoints_then_compacts() { + let dir = tempfile::tempdir().unwrap(); + let env = [("COMMANDS", "compact"), ("USED", "800")]; + let runtime = agent(dir.path(), "ok", &env, policy(75)); + + let done = runtime + .run(&config(dir.path()), "one", &NoopSink) + .await + .unwrap(); + + assert!(done.compacted); + assert_eq!( + prompts(dir.path()), + [ + sent("s1", "SYSTEM PROMPT\n\none"), + sent("s1", "CHECKPOINT"), + sent("s1", "/compact"), + ] + ); + } } diff --git a/hive-runtime/src/acp/stream.rs b/hive-runtime/src/acp/stream.rs index 9460eed4..10d2c503 100644 --- a/hive-runtime/src/acp/stream.rs +++ b/hive-runtime/src/acp/stream.rs @@ -185,6 +185,23 @@ pub(super) fn session_model(response: &Value) -> Option { .map(str::to_owned) } +/// The command names an `available_commands_update` advertises, or `None` +/// for any other `update`. Each replaces the session's previous list. +pub(super) fn advertised_commands(update: &Value) -> Option> { + if update.get("sessionUpdate").and_then(Value::as_str) != Some("available_commands_update") { + return None; + } + let commands = update.get("availableCommands").and_then(Value::as_array); + Some( + commands + .into_iter() + .flatten() + .filter_map(|c| c.get("name").and_then(Value::as_str)) + .map(str::to_owned) + .collect(), + ) +} + /// Convert a claude `--mcp-config` document into ACP's `mcpServers` list. /// Returns the list and the server names, in the same order. pub(super) fn mcp_servers(config: &Value) -> (Vec, Vec) { diff --git a/hive-runtime/src/lib.rs b/hive-runtime/src/lib.rs index 63a3ba1f..35e857b6 100644 --- a/hive-runtime/src/lib.rs +++ b/hive-runtime/src/lib.rs @@ -44,8 +44,8 @@ pub trait Runtime { sink: &impl Sink, ) -> impl Future>; - /// Compact the durable session now. A backend that cannot returns - /// [`Error::Unsupported`]. + /// Compact the durable session now. With no session there is nothing to + /// compact, and this succeeds without doing anything. fn compact(&self, config: &Config, sink: &impl Sink) -> impl Future>; /// Set the durable session aside so the next turn starts a fresh one. @@ -63,7 +63,7 @@ pub trait Runtime { /// [`RuntimeSpec`]. pub enum AgentRuntime { Claude(ClaudeRuntime

), - Acp(Box), + Acp(Box>), } impl Runtime for AgentRuntime

{ @@ -106,9 +106,6 @@ pub enum Error { /// From the ACP backend. #[error(transparent)] Acp(#[from] AcpError), - /// The backend does not implement this operation. - #[error("{0} is not supported by this runtime")] - Unsupported(&'static str), } /// Result alias for this crate.