hive-runtime: shared runtime crate with claude and acp backends
A `Runtime` trait (run / compact / archive) with two backends: - claude: a pass-through to hive_claude's InfiniteSession and SessionStore, so a claude turn is the same spawn, session handling and errors as before. - acp: a generic Agent Client Protocol client. It spawns the command, args and env from RuntimeSpec (HIVE_RUNTIME / HIVE_ACP_COMMAND / HIVE_ACP_ARGS / HIVE_ACP_ENV), refuses an agent whose mcpCapabilities.http is not true, passes the claude --mcp-config servers as ACP mcpServers, keeps one session id in a file (session/load after a restart, session/new otherwise), and maps session/update into claude stream-json events plus usage_update into Telemetry. Permission requests are answered by a caller-supplied policy on the ACP tool kind. compact returns Unsupported for now. The crate depends on no hyperhive binary crate, so the subagent daemon can move onto it without pulling in hive-agent. Refs #4391
This commit is contained in:
atlas
mara
parent
d98d407bf1
commit
b0e26e7e44
10 changed files with 1583 additions and 0 deletions
13
Cargo.lock
generated
13
Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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" }
|
||||
|
|
|
|||
16
hive-runtime/Cargo.toml
Normal file
16
hive-runtime/Cargo.toml
Normal file
|
|
@ -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
|
||||
37
hive-runtime/README.md
Normal file
37
hive-runtime/README.md
Normal file
|
|
@ -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__<server>__<tool>`), 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.
|
||||
365
hive-runtime/src/acp/mod.rs
Normal file
365
hive-runtime/src/acp/mod.rs
Normal file
|
|
@ -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<dyn Fn(&str) -> 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<Option<Live>>,
|
||||
}
|
||||
|
||||
/// The running agent process and the session loaded into it.
|
||||
struct Live {
|
||||
conn: Connection,
|
||||
load_session: bool,
|
||||
session: Option<String>,
|
||||
model: Option<String>,
|
||||
}
|
||||
|
||||
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<Live> {
|
||||
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<bool> {
|
||||
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<Progress> {
|
||||
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<Progress> {
|
||||
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<Option<PathBuf>> {
|
||||
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<String>,
|
||||
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<Value>, Vec<String>)> {
|
||||
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))
|
||||
}
|
||||
316
hive-runtime/src/acp/rpc.rs
Normal file
316
hive-runtime/src/acp/rpc.rs
Normal file
|
|
@ -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<Value, AcpError>;
|
||||
type Pending = Arc<Mutex<HashMap<u64, (&'static str, oneshot::Sender<Reply>)>>>;
|
||||
|
||||
pub(super) struct Connection {
|
||||
child: Child,
|
||||
stdin: Arc<tokio::sync::Mutex<ChildStdin>>,
|
||||
next_id: AtomicU64,
|
||||
pending: Pending,
|
||||
pub(super) incoming: mpsc::UnboundedReceiver<Incoming>,
|
||||
}
|
||||
|
||||
impl Connection {
|
||||
/// Spawn the agent in `cwd` and start reading its output.
|
||||
pub(super) fn spawn(
|
||||
command: &AcpCommand,
|
||||
cwd: &Path,
|
||||
permit: PermissionPolicy,
|
||||
) -> Result<Self, AcpError> {
|
||||
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<oneshot::Receiver<Reply>, 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<Value, AcpError> {
|
||||
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<tokio::sync::Mutex<ChildStdin>>,
|
||||
pending: Pending,
|
||||
tx: mpsc::UnboundedSender<Incoming>,
|
||||
permit: PermissionPolicy,
|
||||
}
|
||||
|
||||
impl Reader {
|
||||
async fn dispatch(&self, line: &str) {
|
||||
let Ok(message) = serde_json::from_str::<Value>(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<ChildStdin>, 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" } })
|
||||
);
|
||||
}
|
||||
}
|
||||
515
hive-runtime/src/acp/stream.rs
Normal file
515
hive-runtime/src/acp/stream.rs
Normal file
|
|
@ -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<String>,
|
||||
text: String,
|
||||
thought: String,
|
||||
tools: HashMap<String, ToolCall>,
|
||||
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__<server>__<tool>` names.
|
||||
pub(super) fn new(servers: Vec<String>) -> 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<Value> {
|
||||
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<Value> {
|
||||
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<Value>) {
|
||||
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<Value>) {
|
||||
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<Value>) {
|
||||
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<String> {
|
||||
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<Value>, Vec<String>) {
|
||||
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 `<server>_<tool>` or
|
||||
/// `<server>__<tool>`: `mcp__<server>__<tool>`, 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<Value> {
|
||||
let mut out: Vec<Value> = 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);
|
||||
}
|
||||
}
|
||||
42
hive-runtime/src/claude.rs
Normal file
42
hive-runtime/src/claude.rs
Normal file
|
|
@ -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<P: CompactionPolicy> {
|
||||
session: InfiniteSession<P>,
|
||||
store: SessionStore,
|
||||
title: String,
|
||||
}
|
||||
|
||||
impl<P: CompactionPolicy> ClaudeRuntime<P> {
|
||||
/// A runtime for the session titled `title`, found through `store` and
|
||||
/// compacted per `policy`.
|
||||
pub fn new(title: impl Into<String>, store: SessionStore, policy: P) -> Self {
|
||||
let title = title.into();
|
||||
Self {
|
||||
session: InfiniteSession::new(title.clone(), store.clone(), policy),
|
||||
store,
|
||||
title,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<P: CompactionPolicy> Runtime for ClaudeRuntime<P> {
|
||||
async fn run(&self, config: &Config, prompt: &str, sink: &impl Sink) -> Result<Progress> {
|
||||
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<Option<PathBuf>> {
|
||||
Ok(self.store.archive_by_title(&self.title)?)
|
||||
}
|
||||
}
|
||||
103
hive-runtime/src/lib.rs
Normal file
103
hive-runtime/src/lib.rs
Normal file
|
|
@ -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<Output = Result<Progress>>;
|
||||
|
||||
/// Compact the durable session now. A backend that cannot returns
|
||||
/// [`Error::Unsupported`].
|
||||
fn compact(&self, config: &Config, sink: &impl Sink) -> impl Future<Output = Result<()>>;
|
||||
|
||||
/// 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<Option<PathBuf>>;
|
||||
}
|
||||
|
||||
/// The runtime an agent was configured with, chosen at startup from a
|
||||
/// [`RuntimeSpec`].
|
||||
pub enum AgentRuntime<P: CompactionPolicy> {
|
||||
Claude(ClaudeRuntime<P>),
|
||||
Acp(Box<AcpRuntime>),
|
||||
}
|
||||
|
||||
impl<P: CompactionPolicy> Runtime for AgentRuntime<P> {
|
||||
async fn run(&self, config: &Config, prompt: &str, sink: &impl Sink) -> Result<Progress> {
|
||||
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<Option<PathBuf>> {
|
||||
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<T> = std::result::Result<T, Error>;
|
||||
174
hive-runtime/src/spec.rs
Normal file
174
hive-runtime/src/spec.rs
Normal file
|
|
@ -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<String>,
|
||||
pub env: BTreeMap<String, String>,
|
||||
}
|
||||
|
||||
/// 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, SpecError> {
|
||||
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<String>) -> Result<Self, SpecError> {
|
||||
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<T: Default + serde::de::DeserializeOwned>(
|
||||
lookup: &impl Fn(&str) -> Option<String>,
|
||||
var: &'static str,
|
||||
) -> Result<T, SpecError> {
|
||||
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<RuntimeSpec, SpecError> {
|
||||
let vars: HashMap<String, String> = 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"
|
||||
));
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue