diff --git a/Cargo.lock b/Cargo.lock index 194abcef..14729ee6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1674,6 +1674,7 @@ dependencies = [ "hive-claude", "hive-core-agent-sock", "hive-log", + "hive-runtime", "hive-sh4re", "hive-sock-client", "http-body-util", @@ -1979,6 +1980,18 @@ dependencies = [ "serde", ] +[[package]] +name = "hive-runtime" +version = "0.1.0" +dependencies = [ + "hive-claude", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tracing", +] + [[package]] name = "hive-screen-mcp" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 6e70491f..8a1ece0f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ members = [ "hive-metric", "hive-priv", "hive-priv-sock", + "hive-runtime", "hive-sh4re", "hive-sock-client", "hive-subagent-mcp", @@ -99,6 +100,7 @@ hive-core-agent-sock = { path = "hive-core-agent-sock" } hive-claude = "0.1.1" hive-host-sock = { path = "hive-host-sock" } hive-priv-sock = { path = "hive-priv-sock" } +hive-runtime = { path = "hive-runtime" } hive-sock-client = { path = "hive-sock-client" } hive-types = { path = "hive-types" } swarm-authelia-bridge-sock = { path = "swarm-authelia-bridge-sock" } diff --git a/hive-runtime/Cargo.toml b/hive-runtime/Cargo.toml new file mode 100644 index 00000000..7039ef75 --- /dev/null +++ b/hive-runtime/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "hive-runtime" +edition.workspace = true +version.workspace = true +readme = "README.md" + +[lints] +workspace = true + +[dependencies] +hive-claude.workspace = true +serde.workspace = true +serde_json.workspace = true +thiserror.workspace = true +tokio.workspace = true +tracing.workspace = true diff --git a/hive-runtime/README.md b/hive-runtime/README.md new file mode 100644 index 00000000..d6cbf8ff --- /dev/null +++ b/hive-runtime/README.md @@ -0,0 +1,37 @@ +# hive-runtime + +The layer an agent's turns are driven through: one `Runtime` interface +(`run`, `compact`, `archive`) with a backend per runtime. + +- **claude** — `claude --print` through the `hive-claude` crate's + `InfiniteSession`. A pass-through: same spawn, same session handling, same + errors. +- **acp** — any [Agent Client Protocol](https://agentclientprotocol.com) + agent, spawned from a command, args and env handed to it (`RuntimeSpec`, + read from `HIVE_RUNTIME` / `HIVE_ACP_COMMAND` / `HIVE_ACP_ARGS` / + `HIVE_ACP_ENV`). It knows no agent by name; which agent runs, and how it is + configured, is decided in nix (`services.hyperhive.agent.runtime`, + `services.hyperhive.agent.acp.*`). + +Both backends report a turn through `hive_claude::Sink` in claude's +`stream-json` shape. The ACP backend translates `session/update` +notifications into it (text and thought chunks as whole blocks, tool calls as +`tool_use` + `tool_result`, MCP tools named `mcp____`), so the +harness's stream consumers read either backend unchanged. Context usage comes +from ACP `usage_update`. + +The crate depends on no hyperhive binary crate, so `hive-agent` and +`hive-subagent-mcp` can both drive turns through it. + +## ACP backend: what it needs from the agent + +- `mcpCapabilities.http` in its `initialize` response. The hyperhive tools + are only served over HTTP, so an agent without it is refused at startup. +- `loadSession`, to pick its session back up after a harness restart. + Without it every restart starts a new session. + +## ACP backend: not yet + +`compact` returns `Error::Unsupported`, there is no cancel, and no idle +watchdog (`Config::idle_timeout` is ignored). An agent that retries a failing +provider on its own keeps the turn open until it gives up. diff --git a/hive-runtime/src/acp/mod.rs b/hive-runtime/src/acp/mod.rs new file mode 100644 index 00000000..223f34a1 --- /dev/null +++ b/hive-runtime/src/acp/mod.rs @@ -0,0 +1,365 @@ +//! The ACP backend: one long-lived agent process per runtime, spawned on the +//! first turn, holding one durable session whose id is kept in a file. + +mod rpc; +mod stream; + +use std::collections::VecDeque; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use hive_claude::{Config, Progress, Sink}; +use serde_json::{Value, json}; +use tokio::sync::Mutex; + +use self::rpc::{Connection, Incoming}; +use self::stream::StreamMapper; +use crate::spec::AcpCommand; +use crate::{Error, Result, Runtime}; + +/// ACP protocol version this client speaks. +const PROTOCOL_VERSION: u64 = 1; + +/// Stderr lines kept for the error when the agent exits mid-turn. +const STDERR_TAIL: usize = 20; + +/// After a turn's reply, how long the agent must stay quiet before the turn +/// is taken to be over, and the most that wait may take in all. +const SETTLE_QUIET: std::time::Duration = std::time::Duration::from_millis(500); +const SETTLE_MAX: std::time::Duration = std::time::Duration::from_secs(5); + +/// Decides the agent's `session/request_permission` requests from the tool +/// call's ACP `kind` (`read`, `edit`, `execute`, `fetch`, …; `other` when the +/// agent gives none). `true` allows the call once, `false` rejects it. +pub type PermissionPolicy = Arc bool + Send + Sync>; + +/// Why an ACP operation failed. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum AcpError { + #[error("failed to spawn ACP agent `{program}`: {source}")] + Spawn { + program: String, + #[source] + source: std::io::Error, + }, + #[error("ACP agent closed its output (exited?)")] + Closed, + #[error("ACP agent exited mid-turn; last stderr:\n{stderr_tail}")] + Exited { stderr_tail: String }, + #[error("ACP agent i/o failed: {0}")] + Io(#[from] std::io::Error), + #[error("ACP `{method}` failed: {message} (code {code}) {data}")] + Rpc { + method: &'static str, + code: i64, + message: String, + data: String, + }, + #[error("ACP `{method}` response has no {field}")] + Malformed { + method: &'static str, + field: &'static str, + }, + #[error( + "the ACP agent does not accept MCP servers over HTTP (`mcpCapabilities.http` is not true), \ + and the hyperhive tools are only served over HTTP" + )] + NoHttpMcp, + #[error("reading the MCP config {path}: {detail}")] + McpConfig { path: PathBuf, detail: String }, +} + +/// Turns on an ACP agent's durable session. +/// +/// From the [`Config`] it reads `cwd` (the agent's working directory), +/// `mcp_config` (converted to the session's `mcpServers`) and +/// `system_prompt_file` (prepended to the first prompt of each new session, +/// since ACP has no system prompt). Everything else in it is claude's and is +/// ignored, including `idle_timeout`. +pub struct AcpRuntime { + command: AcpCommand, + session_file: PathBuf, + permit: PermissionPolicy, + live: Mutex>, +} + +/// The running agent process and the session loaded into it. +struct Live { + conn: Connection, + load_session: bool, + session: Option, + model: Option, +} + +impl AcpRuntime { + /// A runtime spawning `command`, keeping its session id in + /// `session_file`, and answering its permission requests with `permit`. + #[must_use] + pub fn new(command: AcpCommand, session_file: PathBuf, permit: PermissionPolicy) -> Self { + Self { + command, + session_file, + permit, + live: Mutex::new(None), + } + } + + async fn start(&self, cwd: &Path) -> Result { + let conn = Connection::spawn(&self.command, cwd, self.permit.clone())?; + let init = conn + .request( + "initialize", + json!({ + "protocolVersion": PROTOCOL_VERSION, + "clientCapabilities": { + "fs": { "readTextFile": false, "writeTextFile": false }, + "terminal": false, + }, + "clientInfo": { "name": "hyperhive", "version": env!("CARGO_PKG_VERSION") }, + }), + ) + .await?; + let caps = &init["agentCapabilities"]; + if caps["mcpCapabilities"]["http"] != Value::Bool(true) { + return Err(AcpError::NoHttpMcp.into()); + } + tracing::info!(agent = %init["agentInfo"], "ACP agent initialized"); + Ok(Live { + conn, + load_session: caps["loadSession"] == Value::Bool(true), + session: None, + model: None, + }) + } + + /// Attach the durable session: the one already loaded, else the one named + /// by the session file, else a new one. Returns whether it is new. + async fn attach(&self, live: &mut Live, cwd: &Path, servers: &[Value]) -> Result { + let persisted = std::fs::read_to_string(&self.session_file) + .ok() + .map(|s| s.trim().to_owned()) + .filter(|s| !s.is_empty()); + if let Some(id) = persisted { + if live.session.as_deref() == Some(id.as_str()) { + return Ok(false); + } + if live.load_session { + 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.session = Some(id); + return Ok(false); + } + Err(e @ AcpError::Rpc { .. }) => { + tracing::warn!(error = %e, "ACP session/load failed; starting a new session"); + } + Err(e) => return Err(e.into()), + } + } + } + let params = json!({ "cwd": cwd, "mcpServers": servers }); + let response = live.conn.request("session/new", params).await?; + let id = response["sessionId"] + .as_str() + .ok_or(AcpError::Malformed { + method: "session/new", + field: "sessionId", + })? + .to_owned(); + if let Some(dir) = self.session_file.parent() { + std::fs::create_dir_all(dir).map_err(AcpError::Io)?; + } + std::fs::write(&self.session_file, &id).map_err(AcpError::Io)?; + live.model = stream::session_model(&response); + live.session = Some(id); + Ok(true) + } + + async fn turn( + &self, + live: &mut Live, + config: &Config, + prompt: &str, + sink: &impl Sink, + ) -> Result { + let cwd = session_cwd(config); + let (servers, names) = mcp_servers(config)?; + let created = self.attach(live, &cwd, &servers).await?; + let session = live.session.clone().unwrap_or_default(); + let text = match (created, &config.system_prompt_file) { + (true, Some(path)) => match std::fs::read_to_string(path) { + Ok(system) => format!("{system}\n\n{prompt}"), + Err(e) => { + tracing::warn!(path = %path.display(), error = %e, "system prompt unreadable"); + prompt.to_owned() + } + }, + _ => prompt.to_owned(), + }; + let params = json!({ "sessionId": session, "prompt": [{ "type": "text", "text": text }] }); + discard_stale(&mut live.conn)?; + let mut reply = live.conn.send("session/prompt", params).await?; + let mut mapper = StreamMapper::new(names); + let mut stderr_tail = VecDeque::new(); + let response = loop { + tokio::select! { + // Notifications first: the ones sent before the reply belong + // to this turn and must reach the sink before it ends. + biased; + incoming = live.conn.incoming.recv() => { + let incoming = incoming.unwrap_or(Incoming::Closed); + if !deliver(incoming, &session, &mut mapper, &mut stderr_tail, sink) { + let stderr_tail = Vec::from(stderr_tail).join("\n"); + return Err(AcpError::Exited { stderr_tail }.into()); + } + } + reply = &mut reply => break reply.unwrap_or(Err(AcpError::Closed))?, + } + }; + // Updates may still arrive after the reply: an agent can forward its + // stream asynchronously. Take them until the agent goes quiet. + let settled = tokio::time::Instant::now() + SETTLE_MAX; + while let Ok(Some(incoming)) = tokio::time::timeout_at( + settled.min(tokio::time::Instant::now() + SETTLE_QUIET), + live.conn.incoming.recv(), + ) + .await + { + if !deliver(incoming, &session, &mut mapper, &mut stderr_tail, sink) { + break; + } + } + for event in mapper.finish() { + sink.on_event(&event); + } + match response["stopReason"].as_str() { + Some("end_turn") => {} + reason => { + let reason = reason.unwrap_or("none"); + tracing::warn!(reason, "ACP turn stopped early"); + sink.on_stderr_line(&format!("ACP turn stopped: {reason}")); + } + } + let mut progress = Progress::default(); + progress.created = created; + 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() + { + tracing::warn!("ACP agent has exited; respawning"); + *guard = None; + } + let live = match guard.as_mut() { + Some(live) => live, + None => guard.insert(self.start(&session_cwd(config)).await?), + }; + let result = self.turn(live, config, prompt, sink).await; + if let Err(Error::Acp( + AcpError::Closed + | AcpError::Exited { .. } + | AcpError::Io(_) + | AcpError::Malformed { .. }, + )) = &result + { + *guard = None; + } + result + } + + async fn compact(&self, _config: &Config, _sink: &impl Sink) -> Result<()> { + Err(Error::Unsupported("compact")) + } + + /// Moves the session file aside; the agent keeps its own copy of the + /// session, so nothing is lost. + fn archive(&self) -> Result> { + if !self.session_file.exists() { + return Ok(None); + } + let stamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_secs()); + let mut archived = self.session_file.clone().into_os_string(); + archived.push(format!(".{stamp}.archived")); + let archived = PathBuf::from(archived); + std::fs::rename(&self.session_file, &archived).map_err(AcpError::Io)?; + Ok(Some(archived)) + } +} + +/// Hand one incoming item to the sink. Returns `false` once the agent has +/// closed its output. +fn deliver( + incoming: Incoming, + session: &str, + mapper: &mut StreamMapper, + stderr_tail: &mut VecDeque, + sink: &impl Sink, +) -> bool { + match incoming { + Incoming::Update(params) => { + if params["sessionId"].as_str() == Some(session) { + for event in mapper.push(¶ms["update"]) { + sink.on_event(&event); + } + } + } + Incoming::Stdout(line) => sink.on_stdout_line(&line), + Incoming::Stderr(line) => { + sink.on_stderr_line(&line); + if stderr_tail.len() == STDERR_TAIL { + stderr_tail.pop_front(); + } + stderr_tail.push_back(line); + } + Incoming::Closed => return false, + } + true +} + +/// 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), + } + } + Ok(()) +} + +fn session_cwd(config: &Config) -> PathBuf { + config + .cwd + .clone() + .or_else(|| std::env::current_dir().ok()) + .unwrap_or_else(|| PathBuf::from(".")) +} + +fn mcp_servers(config: &Config) -> Result<(Vec, Vec)> { + let Some(path) = &config.mcp_config else { + return Ok((Vec::new(), Vec::new())); + }; + let fail = |detail: String| AcpError::McpConfig { + path: path.clone(), + detail, + }; + let raw = std::fs::read_to_string(path).map_err(|e| fail(e.to_string()))?; + let parsed: Value = serde_json::from_str(&raw).map_err(|e| fail(e.to_string()))?; + Ok(stream::mcp_servers(&parsed)) +} diff --git a/hive-runtime/src/acp/rpc.rs b/hive-runtime/src/acp/rpc.rs new file mode 100644 index 00000000..be265c9a --- /dev/null +++ b/hive-runtime/src/acp/rpc.rs @@ -0,0 +1,316 @@ +//! JSON-RPC 2.0 over an ACP agent's stdio: newline-delimited messages, the +//! agent's requests answered here, its notifications queued for the turn. + +use std::collections::HashMap; +use std::path::Path; +use std::process::Stdio; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, PoisonError}; + +use serde_json::{Value, json}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::process::{Child, ChildStdin, Command}; +use tokio::sync::{mpsc, oneshot}; + +use super::{AcpError, PermissionPolicy}; +use crate::spec::AcpCommand; + +/// Something the agent sent that is not a response to one of our requests. +pub(super) enum Incoming { + /// The `params` of a `session/update` notification. + Update(Value), + Stdout(String), + Stderr(String), + /// The agent closed its stdout: it has exited or is about to. + Closed, +} + +type Reply = Result; +type Pending = Arc)>>>; + +pub(super) struct Connection { + child: Child, + stdin: Arc>, + next_id: AtomicU64, + pending: Pending, + pub(super) incoming: mpsc::UnboundedReceiver, +} + +impl Connection { + /// Spawn the agent in `cwd` and start reading its output. + pub(super) fn spawn( + command: &AcpCommand, + cwd: &Path, + permit: PermissionPolicy, + ) -> Result { + let mut child = Command::new(&command.command) + .args(&command.args) + .envs(&command.env) + .current_dir(cwd) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn() + .map_err(|source| AcpError::Spawn { + program: command.command.clone(), + source, + })?; + let (Some(stdin), Some(stdout), Some(stderr)) = + (child.stdin.take(), child.stdout.take(), child.stderr.take()) + else { + unreachable!("all three stdio handles were requested as pipes"); + }; + let stdin = Arc::new(tokio::sync::Mutex::new(stdin)); + let pending: Pending = Arc::default(); + let (tx, incoming) = mpsc::unbounded_channel(); + + let stderr_tx = tx.clone(); + tokio::spawn(async move { + let mut lines = BufReader::new(stderr).lines(); + while let Ok(Some(line)) = lines.next_line().await { + if stderr_tx.send(Incoming::Stderr(line)).is_err() { + break; + } + } + }); + + let reader = Reader { + stdin: stdin.clone(), + pending: pending.clone(), + tx, + permit, + }; + tokio::spawn(async move { + let mut lines = BufReader::new(stdout).lines(); + while let Ok(Some(line)) = lines.next_line().await { + reader.dispatch(&line).await; + } + // Fail every request still waiting: no reply can come now. + reader + .pending + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clear(); + let _ = reader.tx.send(Incoming::Closed); + }); + + Ok(Self { + child, + stdin, + next_id: AtomicU64::new(1), + pending, + incoming, + }) + } + + /// Whether the agent process has exited. + pub(super) fn exited(&mut self) -> bool { + !matches!(self.child.try_wait(), Ok(None)) + } + + /// Send a request and return the receiver its reply arrives on. + pub(super) async fn send( + &self, + method: &'static str, + params: Value, + ) -> Result, AcpError> { + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let (tx, rx) = oneshot::channel(); + self.pending + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert(id, (method, tx)); + let message = json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params }); + if let Err(e) = write(&self.stdin, &message).await { + self.pending + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(&id); + return Err(e); + } + Ok(rx) + } + + /// Send a request and wait for its reply, queueing anything else the agent + /// sends meanwhile on [`Self::incoming`]. + pub(super) async fn request( + &self, + method: &'static str, + params: Value, + ) -> Result { + let rx = self.send(method, params).await?; + rx.await.unwrap_or(Err(AcpError::Closed)) + } +} + +/// The stdout side: routes responses to their waiting request, answers the +/// agent's own requests, and queues its notifications. +struct Reader { + stdin: Arc>, + pending: Pending, + tx: mpsc::UnboundedSender, + permit: PermissionPolicy, +} + +impl Reader { + async fn dispatch(&self, line: &str) { + let Ok(message) = serde_json::from_str::(line) else { + if !line.trim().is_empty() { + let _ = self.tx.send(Incoming::Stdout(line.to_owned())); + } + return; + }; + let method = message.get("method").and_then(Value::as_str); + let id = message.get("id"); + match (method, id) { + (None, Some(id)) => self.resolve(id, &message), + (Some(method), Some(id)) => { + let reply = match method { + "session/request_permission" => json!({ + "jsonrpc": "2.0", "id": id, + "result": permission_outcome(&message["params"], &self.permit), + }), + _ => json!({ + "jsonrpc": "2.0", "id": id, + "error": { "code": -32601, "message": format!("method not found: {method}") }, + }), + }; + if let Err(e) = write(&self.stdin, &reply).await { + tracing::warn!(error = %e, method, "failed to answer ACP agent request"); + } + } + (Some("session/update"), None) => { + let params = message.get("params").cloned().unwrap_or(Value::Null); + let _ = self.tx.send(Incoming::Update(params)); + } + _ => {} + } + } + + fn resolve(&self, id: &Value, message: &Value) { + let Some(id) = id.as_u64() else { return }; + let Some((method, tx)) = self + .pending + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(&id) + else { + return; + }; + let reply = match message.get("error") { + Some(error) => Err(AcpError::Rpc { + method, + code: error.get("code").and_then(Value::as_i64).unwrap_or(0), + message: error + .get("message") + .and_then(Value::as_str) + .unwrap_or("") + .to_owned(), + data: error.get("data").map(Value::to_string).unwrap_or_default(), + }), + None => Ok(message.get("result").cloned().unwrap_or(Value::Null)), + }; + let _ = tx.send(reply); + } +} + +async fn write(stdin: &tokio::sync::Mutex, message: &Value) -> Result<(), AcpError> { + let mut line = message.to_string(); + line.push('\n'); + let mut stdin = stdin.lock().await; + stdin.write_all(line.as_bytes()).await?; + stdin.flush().await?; + Ok(()) +} + +/// The `session/request_permission` result: the agent's first option of the +/// matching kind, allow if `permit` accepts the tool call's kind and reject +/// otherwise. With no such option the request is answered `cancelled`. +pub(super) fn permission_outcome(params: &Value, permit: &PermissionPolicy) -> Value { + let kind = params + .get("toolCall") + .and_then(|t| t.get("kind")) + .and_then(Value::as_str) + .unwrap_or("other"); + let wanted = if permit(kind) { + ["allow_once", "allow_always"] + } else { + ["reject_once", "reject_always"] + }; + let options = params + .get("options") + .and_then(Value::as_array) + .map_or(&[][..], Vec::as_slice); + let chosen = wanted.iter().find_map(|want| { + options + .iter() + .find(|o| o.get("kind").and_then(Value::as_str) == Some(want)) + .and_then(|o| o.get("optionId")) + }); + match chosen { + Some(option) => json!({ "outcome": { "outcome": "selected", "optionId": option } }), + None => json!({ "outcome": { "outcome": "cancelled" } }), + } +} + +#[cfg(test)] +mod tests { + use super::permission_outcome; + use crate::acp::PermissionPolicy; + use serde_json::json; + use std::sync::Arc; + + fn request(kind: &str) -> serde_json::Value { + json!({ + "sessionId": "s", + "toolCall": { "toolCallId": "t", "kind": kind }, + "options": [ + { "optionId": "once", "kind": "allow_once", "name": "Allow once" }, + { "optionId": "always", "kind": "allow_always", "name": "Always allow" }, + { "optionId": "reject", "kind": "reject_once", "name": "Reject" }, + ], + }) + } + + fn no_execute() -> PermissionPolicy { + Arc::new(|kind: &str| kind != "execute") + } + + #[test] + fn a_permitted_kind_is_allowed_once() { + assert_eq!( + permission_outcome(&request("edit"), &no_execute()), + json!({ "outcome": { "outcome": "selected", "optionId": "once" } }) + ); + } + + #[test] + fn a_refused_kind_is_rejected() { + assert_eq!( + permission_outcome(&request("execute"), &no_execute()), + json!({ "outcome": { "outcome": "selected", "optionId": "reject" } }) + ); + } + + #[test] + fn a_missing_kind_is_judged_as_other() { + let policy: PermissionPolicy = Arc::new(|kind: &str| kind == "other"); + let mut req = request("x"); + req["toolCall"].as_object_mut().unwrap().remove("kind"); + assert_eq!( + permission_outcome(&req, &policy)["outcome"]["optionId"], + "once" + ); + } + + #[test] + fn no_matching_option_cancels() { + let req = json!({ "toolCall": { "kind": "execute" }, + "options": [{ "optionId": "once", "kind": "allow_once" }] }); + assert_eq!( + permission_outcome(&req, &no_execute()), + json!({ "outcome": { "outcome": "cancelled" } }) + ); + } +} diff --git a/hive-runtime/src/acp/stream.rs b/hive-runtime/src/acp/stream.rs new file mode 100644 index 00000000..aadf9067 --- /dev/null +++ b/hive-runtime/src/acp/stream.rs @@ -0,0 +1,515 @@ +//! Pure translation between ACP and the claude-shaped values the rest of the +//! crate's callers read: `session/update` → `stream-json` events, a turn's +//! usage → [`Telemetry`], and a claude `--mcp-config` → ACP `mcpServers`. + +use std::collections::HashMap; + +use hive_claude::{Telemetry, TokenUsage}; +use serde_json::{Map, Value, json}; + +/// Turns one prompt's `session/update` notifications into claude +/// `stream-json` events. +/// +/// Text and thought chunks are buffered and emitted as one block when the +/// update kind changes or the turn ends, so a consumer sees a paragraph where +/// claude would have sent one, not a row per token. A tool call is emitted as +/// a `tool_use` once its input is known, and its `tool_result` when it +/// completes or fails. +pub(super) struct StreamMapper { + servers: Vec, + text: String, + thought: String, + tools: HashMap, + usage: Option<(u64, u64)>, +} + +struct ToolCall { + name: String, + input: Value, + announced: bool, +} + +impl StreamMapper { + /// `servers` are the MCP server names handed to the agent, used to give + /// their tools claude's `mcp____` names. + pub(super) fn new(servers: Vec) -> Self { + let mut servers = servers; + // Longest first, so a server whose name prefixes another's never + // claims the other's tools. + servers.sort_by_key(|s| std::cmp::Reverse(s.len())); + Self { + servers, + text: String::new(), + thought: String::new(), + tools: HashMap::new(), + usage: None, + } + } + + /// Fold one `update` (the `update` field of a `session/update`), returning + /// the events it completes. + pub(super) fn push(&mut self, update: &Value) -> Vec { + let kind = update.get("sessionUpdate").and_then(Value::as_str); + let mut out = Vec::new(); + match kind { + Some("agent_message_chunk") => { + self.flush_thought(&mut out); + self.text.push_str(chunk_text(update)); + } + Some("agent_thought_chunk") => { + self.flush_text(&mut out); + self.thought.push_str(chunk_text(update)); + } + Some("tool_call" | "tool_call_update") => { + self.flush_text(&mut out); + self.flush_thought(&mut out); + self.tool_update(update, &mut out); + } + Some("usage_update") => { + let field = |k: &str| update.get(k).and_then(Value::as_u64); + if let (Some(used), Some(size)) = (field("used"), field("size")) { + self.usage = Some((used, size)); + } + } + _ => {} + } + out + } + + /// Emit whatever is still buffered at the end of the turn. + pub(super) fn finish(&mut self) -> Vec { + let mut out = Vec::new(); + self.flush_text(&mut out); + self.flush_thought(&mut out); + let mut pending: Vec<_> = self.tools.drain().filter(|(_, t)| !t.announced).collect(); + pending.sort_by(|a, b| a.0.cmp(&b.0)); + for (id, tool) in pending { + out.push(tool_use(&id, &tool.name, &tool.input)); + } + out + } + + /// The turn's telemetry: context from the last `usage_update`, cost from + /// the `session/prompt` response's `usage` when the agent sends one. + pub(super) fn telemetry(&self, response: &Value, model: Option<&str>) -> Telemetry { + let mut telemetry = Telemetry::default(); + if let Some((used, size)) = self.usage { + telemetry.context.input_tokens = used; + telemetry.context_window = Some(size); + } + if let Some(usage) = response.get("usage") { + let field = |k: &str| usage.get(k).and_then(Value::as_u64).unwrap_or(0); + let mut cost = TokenUsage::default(); + cost.input_tokens = field("inputTokens"); + cost.output_tokens = field("outputTokens"); + cost.cache_read_input_tokens = field("cachedReadTokens"); + cost.cache_creation_input_tokens = field("cachedWriteTokens"); + telemetry.cost = cost; + } + telemetry.model = model.map(str::to_owned); + telemetry + } + + fn tool_update(&mut self, update: &Value, out: &mut Vec) { + let Some(id) = update.get("toolCallId").and_then(Value::as_str) else { + return; + }; + let tool = self.tools.entry(id.to_owned()).or_insert_with(|| ToolCall { + // The first title is the one kept: later updates may retitle the + // call with a description of what it did. + name: canonical_tool_name( + update + .get("title") + .and_then(Value::as_str) + .unwrap_or("tool"), + &self.servers, + ), + input: json!({}), + announced: false, + }); + if !tool.announced + && let Some(input) = update.get("rawInput").filter(|v| !is_empty(v)) + { + tool.input = input.clone(); + } + let status = update.get("status").and_then(Value::as_str); + let done = matches!(status, Some("completed" | "failed")); + if !tool.announced && (done || !is_empty(&tool.input)) { + tool.announced = true; + out.push(tool_use(id, &tool.name, &tool.input)); + } + if done { + self.tools.remove(id); + out.push(tool_result( + id, + &tool_output(update), + status == Some("failed"), + )); + } + } + + fn flush_text(&mut self, out: &mut Vec) { + if !self.text.is_empty() { + let text = std::mem::take(&mut self.text); + out.push(assistant(json!({ "type": "text", "text": text }))); + } + } + + fn flush_thought(&mut self, out: &mut Vec) { + if !self.thought.is_empty() { + let thinking = std::mem::take(&mut self.thought); + out.push(assistant( + json!({ "type": "thinking", "thinking": thinking }), + )); + } + } +} + +/// The model an agent reports for a session: the current value of its +/// `model`-category config option, else the `models.currentModelId` some +/// agents send instead. +pub(super) fn session_model(response: &Value) -> Option { + let from_options = response + .get("configOptions") + .and_then(Value::as_array) + .and_then(|options| { + options + .iter() + .find(|o| o.get("category").and_then(Value::as_str) == Some("model")) + }) + .and_then(|o| o.get("currentValue")) + .and_then(Value::as_str); + from_options + .or_else(|| { + response + .get("models") + .and_then(|m| m.get("currentModelId")) + .and_then(Value::as_str) + }) + .map(str::to_owned) +} + +/// 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) { + let Some(servers) = config.get("mcpServers").and_then(Value::as_object) else { + return (Vec::new(), Vec::new()); + }; + let mut list = Vec::new(); + let mut names = Vec::new(); + for (name, entry) in servers { + let kind = entry.get("type").and_then(Value::as_str); + let server = match kind { + Some(kind @ ("http" | "sse")) => json!({ + "type": kind, + "name": name, + "url": entry.get("url").cloned().unwrap_or(Value::Null), + "headers": name_values(entry.get("headers")), + }), + _ => json!({ + "name": name, + "command": entry.get("command").cloned().unwrap_or(Value::Null), + "args": entry.get("args").cloned().unwrap_or_else(|| json!([])), + "env": name_values(entry.get("env")), + }), + }; + list.push(server); + names.push(name.clone()); + } + (list, names) +} + +/// The claude name for a tool an ACP agent reports as `_` or +/// `__`: `mcp____`, the form tool-call counters +/// and the terminal renderer key on. Anything else is returned unchanged. +pub(super) fn canonical_tool_name(name: &str, servers: &[String]) -> String { + if name.starts_with("mcp__") { + return name.to_owned(); + } + for server in servers { + if let Some(rest) = name + .strip_prefix(server.as_str()) + .and_then(|r| r.strip_prefix('_')) + { + let tool = rest.strip_prefix('_').unwrap_or(rest); + if !tool.is_empty() { + return format!("mcp__{server}__{tool}"); + } + } + } + name.to_owned() +} + +fn chunk_text(update: &Value) -> &str { + update + .get("content") + .filter(|c| c.get("type").and_then(Value::as_str) == Some("text")) + .and_then(|c| c.get("text")) + .and_then(Value::as_str) + .unwrap_or("") +} + +/// A tool call's output: its text content blocks, else its raw output. +fn tool_output(update: &Value) -> String { + let text: Vec<&str> = update + .get("content") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|c| c.get("content").and_then(|c| c.get("text"))) + .filter_map(Value::as_str) + .collect(); + if !text.is_empty() { + return text.join("\n"); + } + match update.get("rawOutput") { + None | Some(Value::Null) => String::new(), + Some(Value::String(s)) => s.clone(), + Some(other) => other.to_string(), + } +} + +fn is_empty(v: &Value) -> bool { + match v { + Value::Null => true, + Value::Object(m) => m.is_empty(), + _ => false, + } +} + +/// `{"K": "V"}` → `[{"name": "K", "value": "V"}]`, ACP's form for env and +/// headers. +fn name_values(map: Option<&Value>) -> Value { + let pairs = map + .and_then(Value::as_object) + .map(Map::iter) + .into_iter() + .flatten() + .map(|(k, v)| json!({ "name": k, "value": v.as_str().unwrap_or_default() })) + .collect(); + Value::Array(pairs) +} + +fn assistant(block: Value) -> Value { + let mut event = json!({ "type": "assistant", "message": { "role": "assistant" } }); + event["message"]["content"] = Value::Array(vec![block]); + event +} + +fn tool_use(id: &str, name: &str, input: &Value) -> Value { + assistant(json!({ "type": "tool_use", "id": id, "name": name, "input": input })) +} + +fn tool_result(id: &str, output: &str, is_error: bool) -> Value { + json!({ + "type": "user", + "message": { "role": "user", "content": [{ + "type": "tool_result", "tool_use_id": id, "content": output, "is_error": is_error, + }]}, + }) +} + +#[cfg(test)] +mod tests { + use super::{StreamMapper, canonical_tool_name, mcp_servers, session_model}; + use serde_json::{Value, json}; + + fn mapper() -> StreamMapper { + StreamMapper::new(vec!["hyperhive".into(), "bash".into()]) + } + + fn feed(m: &mut StreamMapper, updates: &[Value]) -> Vec { + let mut out: Vec = updates.iter().flat_map(|u| m.push(u)).collect(); + out.extend(m.finish()); + out + } + + fn text_chunk(kind: &str, text: &str) -> Value { + json!({ "sessionUpdate": kind, "content": { "type": "text", "text": text } }) + } + + #[test] + fn message_chunks_become_one_assistant_text_block() { + let out = feed( + &mut mapper(), + &[ + text_chunk("agent_message_chunk", "Hel"), + text_chunk("agent_message_chunk", "lo"), + ], + ); + assert_eq!( + out, + vec![ + json!({ "type": "assistant", "message": { "role": "assistant", + "content": [{ "type": "text", "text": "Hello" }] } }) + ] + ); + } + + #[test] + fn a_thought_then_text_flushes_in_order() { + let out = feed( + &mut mapper(), + &[ + text_chunk("agent_thought_chunk", "hmm"), + text_chunk("agent_message_chunk", "done"), + ], + ); + let blocks: Vec<&Value> = out.iter().map(|e| &e["message"]["content"][0]).collect(); + assert_eq!(blocks[0], &json!({ "type": "thinking", "thinking": "hmm" })); + assert_eq!(blocks[1], &json!({ "type": "text", "text": "done" })); + } + + /// The sequence an agent sends for one MCP tool call: announced with no + /// input, updated with the input while running, then completed with its + /// output and a new, descriptive title. + #[test] + fn a_tool_call_becomes_tool_use_then_tool_result() { + let out = feed( + &mut mapper(), + &[ + text_chunk("agent_message_chunk", "sending"), + json!({ "sessionUpdate": "tool_call", "toolCallId": "c1", + "title": "hyperhive_send", "kind": "other", "status": "pending", + "rawInput": {} }), + json!({ "sessionUpdate": "tool_call_update", "toolCallId": "c1", + "status": "in_progress", "rawInput": { "to": "atlas", "body": "hi" } }), + json!({ "sessionUpdate": "tool_call_update", "toolCallId": "c1", + "status": "completed", "title": "Sent a message", + "content": [{ "type": "content", "content": { "type": "text", "text": "ok" } }] }), + ], + ); + assert_eq!(out.len(), 3, "{out:#?}"); + assert_eq!(out[0]["message"]["content"][0]["text"], "sending"); + assert_eq!( + out[1], + json!({ "type": "assistant", "message": { "role": "assistant", "content": [{ + "type": "tool_use", "id": "c1", "name": "mcp__hyperhive__send", + "input": { "to": "atlas", "body": "hi" } }] } }) + ); + assert_eq!( + out[2], + json!({ "type": "user", "message": { "role": "user", "content": [{ + "type": "tool_result", "tool_use_id": "c1", "content": "ok", "is_error": false }] } }) + ); + } + + #[test] + fn a_failed_tool_call_is_an_error_result_carrying_raw_output() { + let out = feed( + &mut mapper(), + &[ + json!({ "sessionUpdate": "tool_call_update", "toolCallId": "c2", + "title": "read", "status": "failed", + "rawOutput": { "error": "no such file" } }), + ], + ); + assert_eq!(out[0]["message"]["content"][0]["name"], "read"); + assert_eq!(out[1]["message"]["content"][0]["is_error"], true); + assert_eq!( + out[1]["message"]["content"][0]["content"], + r#"{"error":"no such file"}"# + ); + } + + #[test] + fn a_tool_call_still_running_at_turn_end_is_still_counted() { + let out = feed( + &mut mapper(), + &[json!({ "sessionUpdate": "tool_call", "toolCallId": "c3", + "title": "bash_run", "status": "pending" })], + ); + assert_eq!(out.len(), 1); + assert_eq!(out[0]["message"]["content"][0]["name"], "mcp__bash__run"); + } + + #[test] + fn usage_update_and_prompt_usage_become_telemetry() { + let mut m = mapper(); + feed( + &mut m, + &[ + json!({ "sessionUpdate": "usage_update", "used": 1000, "size": 262_144 }), + json!({ "sessionUpdate": "usage_update", "used": 5000, "size": 262_144, + "cost": { "amount": 0.0, "currency": "USD" } }), + ], + ); + let response = json!({ "stopReason": "end_turn", "usage": { + "inputTokens": 4000, "outputTokens": 300, "cachedReadTokens": 1000 } }); + let t = m.telemetry(&response, Some("provider/model")); + assert_eq!(t.context.context_tokens(), 5000); + assert_eq!(t.context_window, Some(262_144)); + assert_eq!(t.cost.input_tokens, 4000); + assert_eq!(t.cost.output_tokens, 300); + assert_eq!(t.cost.cache_read_input_tokens, 1000); + assert_eq!(t.model.as_deref(), Some("provider/model")); + } + + #[test] + fn updates_with_nothing_to_show_emit_nothing() { + let out = feed( + &mut mapper(), + &[ + json!({ "sessionUpdate": "available_commands_update", "availableCommands": [] }), + json!({ "sessionUpdate": "plan", "entries": [] }), + json!({ "sessionUpdate": "something_new" }), + ], + ); + assert!(out.is_empty(), "{out:#?}"); + } + + #[test] + fn tool_names_are_canonicalised_only_for_known_servers() { + let servers = vec!["hyperhive".to_owned(), "bash".to_owned()]; + assert_eq!( + canonical_tool_name("hyperhive_get_loose_ends", &servers), + "mcp__hyperhive__get_loose_ends" + ); + assert_eq!( + canonical_tool_name("hyperhive__send", &servers), + "mcp__hyperhive__send" + ); + assert_eq!( + canonical_tool_name("mcp__bash__run", &servers), + "mcp__bash__run" + ); + assert_eq!(canonical_tool_name("edit", &servers), "edit"); + assert_eq!(canonical_tool_name("bashful", &servers), "bashful"); + assert_eq!(canonical_tool_name("bash_", &servers), "bash_"); + } + + #[test] + fn claude_mcp_config_becomes_acp_mcp_servers() { + let config = json!({ "mcpServers": { + "hyperhive": { "type": "http", "url": "http://127.0.0.1:8790/mcp" }, + "scraper": { "command": "/bin/scrape", "args": ["--x"], + "env": { "HYPERHIVE_STATE_DIR": "/state" } }, + }}); + let (servers, names) = mcp_servers(&config); + assert_eq!(names, vec!["hyperhive", "scraper"]); + assert_eq!( + servers, + vec![ + json!({ "type": "http", "name": "hyperhive", + "url": "http://127.0.0.1:8790/mcp", "headers": [] }), + json!({ "name": "scraper", "command": "/bin/scrape", "args": ["--x"], + "env": [{ "name": "HYPERHIVE_STATE_DIR", "value": "/state" }] }), + ] + ); + } + + #[test] + fn session_model_prefers_the_model_config_option() { + let resp = json!({ + "configOptions": [ + { "id": "mode", "category": "mode", "currentValue": "build" }, + { "id": "model", "category": "model", "currentValue": "p/m-1" }, + ], + "models": { "currentModelId": "p/m-2" }, + }); + assert_eq!(session_model(&resp).as_deref(), Some("p/m-1")); + let resp = json!({ "models": { "currentModelId": "p/m-2" } }); + assert_eq!(session_model(&resp).as_deref(), Some("p/m-2")); + assert_eq!(session_model(&json!({})), None); + } +} diff --git a/hive-runtime/src/claude.rs b/hive-runtime/src/claude.rs new file mode 100644 index 00000000..c6690069 --- /dev/null +++ b/hive-runtime/src/claude.rs @@ -0,0 +1,42 @@ +//! The claude backend: a pass-through to `hive_claude`. + +use std::path::PathBuf; + +use hive_claude::{CompactionPolicy, Config, InfiniteSession, Progress, SessionStore, Sink}; + +use crate::{Result, Runtime}; + +/// `claude --print` turns on a titled [`InfiniteSession`], which owns +/// resume-or-create and compaction. +pub struct ClaudeRuntime { + session: InfiniteSession

, + store: SessionStore, + title: String, +} + +impl ClaudeRuntime

{ + /// A runtime for the session titled `title`, found through `store` and + /// compacted per `policy`. + pub fn new(title: impl Into, store: SessionStore, policy: P) -> Self { + let title = title.into(); + Self { + session: InfiniteSession::new(title.clone(), store.clone(), policy), + store, + title, + } + } +} + +impl Runtime for ClaudeRuntime

{ + async fn run(&self, config: &Config, prompt: &str, sink: &impl Sink) -> Result { + Ok(self.session.run(config, prompt, sink).await?) + } + + async fn compact(&self, config: &Config, sink: &impl Sink) -> Result<()> { + Ok(self.session.compact(config, sink).await?) + } + + fn archive(&self) -> Result> { + Ok(self.store.archive_by_title(&self.title)?) + } +} diff --git a/hive-runtime/src/lib.rs b/hive-runtime/src/lib.rs new file mode 100644 index 00000000..f9142a17 --- /dev/null +++ b/hive-runtime/src/lib.rs @@ -0,0 +1,103 @@ +//! `hive-runtime` — the layer an agent's turns are driven through, with one +//! backend per runtime: +//! +//! - [`ClaudeRuntime`] — `claude --print`, via `hive_claude`'s +//! [`hive_claude::InfiniteSession`]; +//! - [`AcpRuntime`] — any Agent Client Protocol agent, spawned from the +//! command, args and env it is handed (see [`RuntimeSpec`]). It knows no +//! agent by name. +//! +//! Both report a turn through the same [`Sink`], in claude's `stream-json` +//! shape: the ACP backend translates its `session/update` notifications into +//! it, so a consumer of that stream reads either backend unchanged. +//! +//! Depends on no hyperhive binary crate, so the harness and the subagent +//! daemon can both drive turns through it. + +use std::future::Future; +use std::path::PathBuf; + +mod acp; +mod claude; +mod spec; + +pub use acp::{AcpError, AcpRuntime, PermissionPolicy}; +pub use claude::ClaudeRuntime; +pub use hive_claude::{ + CompactionPolicy, Config, PercentPolicy, Progress, SessionStore, Sink, Telemetry, TokenUsage, +}; +pub use spec::{ + ACP_ARGS_ENV, ACP_COMMAND_ENV, ACP_ENV_ENV, AcpCommand, RUNTIME_ENV, RuntimeSpec, SpecError, +}; + +/// The operations every runtime backend offers. +/// +/// Each takes the same [`Config`] the claude backend spawns from; a backend +/// reads the fields that mean something to it and ignores the rest. +pub trait Runtime { + /// Run one turn on the agent's durable session, creating it if there is + /// none, and stream it to `sink`. + fn run( + &self, + config: &Config, + prompt: &str, + sink: &impl Sink, + ) -> impl Future>; + + /// Compact the durable session now. A backend that cannot returns + /// [`Error::Unsupported`]. + fn compact(&self, config: &Config, sink: &impl Sink) -> impl Future>; + + /// Set the durable session aside so the next turn starts a fresh one. + /// Returns the file the session was moved to, or `None` if there was no + /// session to archive. Only call it between turns. + fn archive(&self) -> Result>; +} + +/// The runtime an agent was configured with, chosen at startup from a +/// [`RuntimeSpec`]. +pub enum AgentRuntime { + Claude(ClaudeRuntime

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

{ + async fn run(&self, config: &Config, prompt: &str, sink: &impl Sink) -> Result { + match self { + Self::Claude(r) => r.run(config, prompt, sink).await, + Self::Acp(r) => r.run(config, prompt, sink).await, + } + } + + async fn compact(&self, config: &Config, sink: &impl Sink) -> Result<()> { + match self { + Self::Claude(r) => r.compact(config, sink).await, + Self::Acp(r) => r.compact(config, sink).await, + } + } + + fn archive(&self) -> Result> { + match self { + Self::Claude(r) => r.archive(), + Self::Acp(r) => r.archive(), + } + } +} + +/// Why a runtime operation did not complete. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum Error { + /// From the claude backend, unchanged. + #[error(transparent)] + Claude(#[from] hive_claude::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. +pub type Result = std::result::Result; diff --git a/hive-runtime/src/spec.rs b/hive-runtime/src/spec.rs new file mode 100644 index 00000000..d3ad48f4 --- /dev/null +++ b/hive-runtime/src/spec.rs @@ -0,0 +1,174 @@ +//! Which runtime an agent runs, as read from its environment. + +use std::collections::BTreeMap; + +/// Selects the runtime: `claude` (also when unset) or `acp`. +pub const RUNTIME_ENV: &str = "HIVE_RUNTIME"; +/// The ACP agent's program, spawned as-is (a path or a name on `PATH`). +pub const ACP_COMMAND_ENV: &str = "HIVE_ACP_COMMAND"; +/// The ACP agent's arguments, as a JSON array of strings. Optional. +pub const ACP_ARGS_ENV: &str = "HIVE_ACP_ARGS"; +/// Extra environment for the ACP agent only, as a JSON object of strings. +/// Optional. The agent also inherits the harness's own environment. +pub const ACP_ENV_ENV: &str = "HIVE_ACP_ENV"; + +/// The runtime an agent is configured with. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RuntimeSpec { + Claude, + Acp(AcpCommand), +} + +/// How to spawn an ACP agent. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AcpCommand { + pub command: String, + pub args: Vec, + pub env: BTreeMap, +} + +/// A runtime configuration that cannot be acted on. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum SpecError { + #[error("{RUNTIME_ENV}={0:?} names no runtime (expected \"claude\" or \"acp\")")] + UnknownRuntime(String), + #[error("{RUNTIME_ENV}=acp but {ACP_COMMAND_ENV} is unset or empty")] + MissingCommand, + #[error("{var} is not valid JSON of the expected shape: {source}")] + Malformed { + var: &'static str, + #[source] + source: serde_json::Error, + }, +} + +impl RuntimeSpec { + /// Read the spec from the process environment. + pub fn from_env() -> Result { + Self::parse(|var| std::env::var(var).ok()) + } + + /// Build the spec from `lookup`, which returns an env var's value. + pub fn parse(lookup: impl Fn(&str) -> Option) -> Result { + let runtime = lookup(RUNTIME_ENV).unwrap_or_default(); + match runtime.trim() { + "" | "claude" => Ok(Self::Claude), + "acp" => { + let command = lookup(ACP_COMMAND_ENV).unwrap_or_default(); + if command.trim().is_empty() { + return Err(SpecError::MissingCommand); + } + Ok(Self::Acp(AcpCommand { + command, + args: json_or_default(&lookup, ACP_ARGS_ENV)?, + env: json_or_default(&lookup, ACP_ENV_ENV)?, + })) + } + other => Err(SpecError::UnknownRuntime(other.to_owned())), + } + } +} + +fn json_or_default( + lookup: &impl Fn(&str) -> Option, + var: &'static str, +) -> Result { + match lookup(var) { + Some(raw) if !raw.trim().is_empty() => { + serde_json::from_str(&raw).map_err(|source| SpecError::Malformed { var, source }) + } + _ => Ok(T::default()), + } +} + +#[cfg(test)] +mod tests { + use super::{AcpCommand, RuntimeSpec, SpecError}; + use std::collections::HashMap; + + fn parse(vars: &[(&str, &str)]) -> Result { + let vars: HashMap = vars + .iter() + .map(|(k, v)| ((*k).to_owned(), (*v).to_owned())) + .collect(); + RuntimeSpec::parse(|k| vars.get(k).cloned()) + } + + #[test] + fn unset_or_claude_is_the_claude_runtime() { + assert_eq!(parse(&[]).unwrap(), RuntimeSpec::Claude); + assert_eq!( + parse(&[("HIVE_RUNTIME", "claude")]).unwrap(), + RuntimeSpec::Claude + ); + // ACP settings alone do not switch the runtime. + assert_eq!( + parse(&[("HIVE_ACP_COMMAND", "/bin/agent")]).unwrap(), + RuntimeSpec::Claude + ); + } + + #[test] + fn acp_resolves_command_args_and_env() { + let spec = parse(&[ + ("HIVE_RUNTIME", "acp"), + ("HIVE_ACP_COMMAND", "/nix/store/x/bin/agent"), + ("HIVE_ACP_ARGS", r#"["acp","--flag"]"#), + ( + "HIVE_ACP_ENV", + r#"{"AGENT_CONFIG":"/nix/store/y/config.json"}"#, + ), + ]) + .unwrap(); + assert_eq!( + spec, + RuntimeSpec::Acp(AcpCommand { + command: "/nix/store/x/bin/agent".into(), + args: vec!["acp".into(), "--flag".into()], + env: [("AGENT_CONFIG".into(), "/nix/store/y/config.json".into())].into(), + }) + ); + } + + #[test] + fn acp_args_and_env_are_optional() { + let spec = parse(&[("HIVE_RUNTIME", "acp"), ("HIVE_ACP_COMMAND", "agent")]).unwrap(); + let RuntimeSpec::Acp(cmd) = spec else { + panic!("expected acp, got {spec:?}"); + }; + assert!(cmd.args.is_empty()); + assert!(cmd.env.is_empty()); + } + + #[test] + fn acp_without_a_command_is_refused() { + assert!(matches!( + parse(&[("HIVE_RUNTIME", "acp")]), + Err(SpecError::MissingCommand) + )); + assert!(matches!( + parse(&[("HIVE_RUNTIME", "acp"), ("HIVE_ACP_COMMAND", " ")]), + Err(SpecError::MissingCommand) + )); + } + + #[test] + fn malformed_json_names_the_variable() { + let err = parse(&[ + ("HIVE_RUNTIME", "acp"), + ("HIVE_ACP_COMMAND", "agent"), + ("HIVE_ACP_ARGS", r#"{"not":"a list"}"#), + ]) + .unwrap_err(); + assert!(err.to_string().starts_with("HIVE_ACP_ARGS "), "{err}"); + } + + #[test] + fn an_unknown_runtime_is_refused_rather_than_defaulted() { + assert!(matches!( + parse(&[("HIVE_RUNTIME", "bogus")]), + Err(SpecError::UnknownRuntime(r)) if r == "bogus" + )); + } +}