From 7699db6500a9369158306e7a2facd893708bcbf7 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 9 Sep 2026 18:15:28 +0200 Subject: [PATCH] hive-subagent-mcp: new crate for the subagent daemon, independent of hive-bash-mcp --- Cargo.lock | 36 +++- Cargo.toml | 3 +- hive-subagent-mcp/Cargo.toml | 32 ++++ hive-subagent-mcp/README.md | 27 +++ hive-subagent-mcp/src/lib.rs | 14 ++ hive-subagent-mcp/src/main.rs | 52 +++++ hive-subagent-mcp/src/mcp.rs | 159 ++++++++++++++++ hive-subagent-mcp/src/paths.rs | 15 ++ hive-subagent-mcp/src/session.rs | 317 +++++++++++++++++++++++++++++++ 9 files changed, 646 insertions(+), 9 deletions(-) create mode 100644 hive-subagent-mcp/Cargo.toml create mode 100644 hive-subagent-mcp/README.md create mode 100644 hive-subagent-mcp/src/lib.rs create mode 100644 hive-subagent-mcp/src/main.rs create mode 100644 hive-subagent-mcp/src/mcp.rs create mode 100644 hive-subagent-mcp/src/paths.rs create mode 100644 hive-subagent-mcp/src/session.rs diff --git a/Cargo.lock b/Cargo.lock index bd738290..8563a611 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1256,7 +1256,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -1781,9 +1781,9 @@ dependencies = [ [[package]] name = "hive-claude" version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d540ef2338fc071b60e69cac345cb5ec04627c4ef11c49138d1c39d4f687f254" +source = "git+https://forge.darkest.space/hyperhive/hive-claude?branch=main#bacccc6580504d548387ffff7f0567be603e73a4" dependencies = [ + "libc", "serde", "serde_json", "thiserror 2.0.18", @@ -1976,6 +1976,26 @@ dependencies = [ "tracing", ] +[[package]] +name = "hive-subagent-mcp" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum", + "clap", + "hive-agent-sock", + "hive-claude", + "hive-sock-client", + "hive-types", + "rmcp", + "schemars", + "serde", + "serde_json", + "tokio", + "tracing", + "tracing-subscriber", +] + [[package]] name = "hive-types" version = "0.1.0" @@ -3594,7 +3614,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4123,7 +4143,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4182,7 +4202,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4863,7 +4883,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -5731,7 +5751,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 8ca89bd1..ba738ac0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,6 +20,7 @@ members = [ "hive-priv-sock", "hive-sh4re", "hive-sock-client", + "hive-subagent-mcp", "hive-types", "hivectl", "swarm-authelia-bridge", @@ -84,7 +85,7 @@ hive-jobq = { path = "hive-jobq" } hive-jobq-metrics = { path = "hive-jobq-metrics" } hive-jobq-wire = { path = "hive-jobq-wire" } hive-core-agent-sock = { path = "hive-core-agent-sock" } -hive-claude = "0.1" +hive-claude = { git = "https://forge.darkest.space/hyperhive/hive-claude", branch = "main" } hive-host-sock = { path = "hive-host-sock" } hive-priv-sock = { path = "hive-priv-sock" } hive-sock-client = { path = "hive-sock-client" } diff --git a/hive-subagent-mcp/Cargo.toml b/hive-subagent-mcp/Cargo.toml new file mode 100644 index 00000000..b3315719 --- /dev/null +++ b/hive-subagent-mcp/Cargo.toml @@ -0,0 +1,32 @@ +[package] +name = "hive-subagent-mcp" +edition.workspace = true +version.workspace = true +readme = "README.md" + +[lints] +workspace = true + +[dependencies] +anyhow.workspace = true +axum.workspace = true +clap.workspace = true +hive-agent-sock.workspace = true +hive-claude.workspace = true +hive-sock-client.workspace = true +hive-types.workspace = true +rmcp.workspace = true +schemars.workspace = true +serde.workspace = true +serde_json.workspace = true +tokio.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true + +# `hive-subagent-daemon` — long-running per-agent claude-subagent runner. +# Independent of `hive-bash-mcp` (own crate, own binary, own MCP server) — +# see lib.rs's module doc for why. Serves its MCP tools (`start`/ +# `continue`/`interrupt`) directly over streamable-http — no stdio bridge. +[[bin]] +name = "hive-subagent-daemon" +path = "src/main.rs" diff --git a/hive-subagent-mcp/README.md b/hive-subagent-mcp/README.md new file mode 100644 index 00000000..27ae605b --- /dev/null +++ b/hive-subagent-mcp/README.md @@ -0,0 +1,27 @@ +# hive-subagent-mcp + +Per-agent daemon (`hive-subagent-daemon`) that spawns nested headless +`claude` sessions on request and serves the tool surface +(`start`/`continue`/`interrupt`) directly over streamable-http. No +stdio bridge, no per-turn respawn — claude reconnects to the same +stable URL every turn. + +Independent of `hive-bash-mcp` — a subagent spawns a full nested +`claude` session, a much heavier capability than a bash command, worth +its own deployable/restartable unit. + +## Shape + +One bin (`hive-subagent-daemon`, `src/main.rs`) built from the crate's +own lib (`src/lib.rs`): + +- **`session.rs`** — the actual claude-facing logic: `Claude::spawn` + + `RunningClaude::wait`/`cancel_handle` (not `InfiniteSession::run`, + which has no cancel handle to reach in — see the module doc for the + v1 scope this trades away), the in-memory `running` map that's the + _only_ state this daemon keeps (no task files — a restart stops + whatever's running; the actual claude session is the durable store, + found again by name via `hive_claude::SessionStore`). +- **`mcp.rs`** — the `rmcp` tool router (`start`/`continue`/`interrupt`) + - `serve_http`. +- **`paths.rs`** — the in-agent todo-socket path. diff --git a/hive-subagent-mcp/src/lib.rs b/hive-subagent-mcp/src/lib.rs new file mode 100644 index 00000000..180efee8 --- /dev/null +++ b/hive-subagent-mcp/src/lib.rs @@ -0,0 +1,14 @@ +//! Library for `hive-subagent-daemon`: spawns nested claude sessions on +//! request and serves the `start`/`continue`/`interrupt` MCP tool surface +//! directly over streamable-http — no stdio bridge, no round-trip socket. +//! Independent of `hive-bash-mcp` — a subagent is a much heavier capability +//! than a bash command (a full nested `claude` process), worth its own +//! deployable/restartable unit rather than sharing one. +//! +//! See [`session`]'s module doc for the actual design: no task files, no +//! restart recovery, no mid-turn compaction — the daemon's only state is an +//! in-memory `name -> Cancel` map, live only as long as the process is. + +pub mod mcp; +pub mod paths; +pub mod session; diff --git a/hive-subagent-mcp/src/main.rs b/hive-subagent-mcp/src/main.rs new file mode 100644 index 00000000..40f5fdc3 --- /dev/null +++ b/hive-subagent-mcp/src/main.rs @@ -0,0 +1,52 @@ +//! `hive-subagent-daemon` binary — spawns nested claude sessions on +//! request and serves the `start`/`continue`/`interrupt` MCP tool surface +//! directly over streamable-http on `--http ` — no stdio bridge, no +//! separate bin claude has to respawn every turn. + +use std::sync::Arc; + +use anyhow::Result; +use clap::Parser; + +#[derive(Parser)] +#[command( + name = "hive-subagent-daemon", + about = "claude-subagent runner + MCP daemon" +)] +struct Cli { + /// Serve the MCP tools over streamable-http on this address (e.g. + /// `127.0.0.1:8793`). Bind loopback only. + #[arg(long)] + http: std::net::SocketAddr, +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_env("RUST_LOG") + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + // This is a systemd-managed daemon — stdout always goes to journald, + // never a human terminal, and journald doesn't strip ANSI escapes: + // they land in victorialogs as raw byte-array spam otherwise. + .with_ansi(false) + .init(); + + let cli = Cli::parse(); + let todo_socket = hive_subagent_mcp::paths::agent_socket(); + + tracing::info!( + http = %cli.http, + todo = %todo_socket.display(), + "hive-subagent-daemon starting" + ); + + let state = Arc::new(hive_subagent_mcp::session::State::new(todo_socket)); + + // Serve the MCP tools over streamable-http forever. No background poll + // loop to start — unlike the bash daemon's task-file queue, `start`/ + // `continue` spawn their subagent's background turn directly from the + // tool call itself, nothing to scan for. + hive_subagent_mcp::mcp::serve_http(cli.http, state).await +} diff --git a/hive-subagent-mcp/src/mcp.rs b/hive-subagent-mcp/src/mcp.rs new file mode 100644 index 00000000..493948b3 --- /dev/null +++ b/hive-subagent-mcp/src/mcp.rs @@ -0,0 +1,159 @@ +//! The MCP tool surface: `start` / `continue` / `interrupt`, served +//! directly over streamable-http — no stdio bridge, no round-trip socket. + +use std::sync::Arc; + +use rmcp::{ + ServerHandler, + handler::server::wrapper::Parameters, + schemars::{self, JsonSchema}, + tool, tool_handler, tool_router, +}; +use serde::Deserialize; + +use crate::session::{self, State}; + +#[derive(Debug, Deserialize, JsonSchema)] +struct StartArgs { + /// Session name — becomes both this daemon's tracking key and claude's + /// own `--name`/`--resume` session title. Same identifier rules as the + /// `bash` server's task names: lowercase, digits, hyphen, max 63 chars. + /// Reusable once a prior *finished* session under that name is done — + /// rejected while one under the same name is still running. + name: String, + /// `--model` for the subagent's own claude invocation. Omit for + /// claude's own default. The `base:claude-subagents` skill's + /// "cheaper-than-you" guidance still applies here. + #[serde(default)] + model: Option, + /// Path to a file passed as `--append-system-prompt-file` — the + /// subagent's actual task instructions. A file, not an inline string, + /// to avoid `ARG_MAX` on a large recipe. + prompt_file: String, + /// Written to the subagent's stdin as its first turn's prompt. Default: + /// a generic "carry out your instructions" nudge — the real task detail + /// belongs in `prompt_file`, not here. + #[serde(default = "default_trigger")] + trigger: String, +} + +fn default_trigger() -> String { + "Carry out the task described in your instructions.".to_owned() +} + +#[derive(Debug, Deserialize, JsonSchema)] +struct ContinueArgs { + /// The existing session's name (from a prior `start`). + name: String, + /// The new turn's prompt, written to the subagent's stdin. + prompt: String, + /// `--model` for this turn. Omit to let claude fall back to its own + /// default — this does not have to match whatever model `start` used. + #[serde(default)] + model: Option, +} + +#[derive(Debug, Deserialize, JsonSchema)] +struct InterruptArgs { + /// The running session's name to signal. + name: String, + /// `true` sends SIGKILL immediately; `false` (default) sends SIGINT, + /// letting claude shut down cleanly if it's already mid-response. + #[serde(default)] + force: bool, +} + +#[derive(Clone)] +struct SubagentMcp { + state: Arc, +} + +#[tool_router] +impl SubagentMcp { + #[tool( + description = "Start a fresh claude subagent session under `name`, running in the \ + background. Returns as soon as the process is confirmed running — not once it \ + finishes; poll for completion via the todo this daemon pushes when the turn ends, \ + or use `continue` later to give it another turn. A prior *finished* session under \ + the same name is archived first (real fresh start, not a silent resume); a \ + *currently running* one is refused. Always runs with \ + `--dangerously-skip-permissions --strict-mcp-config` (no `--mcp-config` override — \ + that's a safety property, not a knob). See the `base:claude-subagents` skill for \ + when to reach for this." + )] + fn start(&self, Parameters(args): Parameters) -> String { + match session::start( + &self.state, + &args.name, + args.model, + &args.prompt_file, + args.trigger, + ) { + Ok(msg) => msg, + Err(e) => format!("start error: {e:#}"), + } + } + + #[tool( + name = "continue", + description = "Give an existing named subagent session a new turn — whether that's \ + because its previous turn finished and you have a follow-up instruction, or you're \ + reattaching after this daemon restarted (the session itself survives independently \ + of the daemon that spawned it). Returns as soon as confirmed running, same as \ + `start`. Refuses a name with no session on disk at all, or one already running." + )] + fn r#continue(&self, Parameters(args): Parameters) -> String { + match session::continue_(&self.state, &args.name, args.prompt, args.model) { + Ok(msg) => msg, + Err(e) => format!("continue error: {e:#}"), + } + } + + #[tool( + description = "Signal a currently-running subagent session to stop. Only works while \ + it's actually running — there's no queued/pending state to cancel pre-emptively, \ + only running or not tracked at all. `force: true` for SIGKILL, otherwise SIGINT." + )] + fn interrupt(&self, Parameters(args): Parameters) -> String { + match session::interrupt(&self.state, &args.name, args.force) { + Ok(msg) => msg, + Err(e) => format!("interrupt error: {e:#}"), + } + } +} + +#[tool_handler] +impl ServerHandler for SubagentMcp {} + +/// Run the MCP server over HTTP (rmcp streamable-http transport) on `addr`. +/// Loopback-only bind, one long-lived session — same shape as the bash and +/// matrix daemons' own `serve_http`. +/// +/// # Errors +/// +/// Returns an error if the listener cannot bind `addr` or the HTTP server +/// exits with a fatal error. +pub async fn serve_http(addr: std::net::SocketAddr, state: Arc) -> anyhow::Result<()> { + use rmcp::transport::streamable_http_server::{ + StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, + }; + let mut session_manager = LocalSessionManager::default(); + // A subagent turn can run considerably longer than a bash command — + // same 24h keep-alive rationale as the bash/matrix daemons. + session_manager.session_config.keep_alive = Some(std::time::Duration::from_hours(24)); + let session_manager = std::sync::Arc::new(session_manager); + let service = StreamableHttpService::new( + move || { + Ok(SubagentMcp { + state: Arc::clone(&state), + }) + }, + session_manager, + StreamableHttpServerConfig::default(), + ); + let app = axum::Router::new().nest_service("/mcp", service); + let listener = tokio::net::TcpListener::bind(addr).await?; + tracing::info!(%addr, "serving hive-subagent MCP over streamable-http at /mcp"); + axum::serve(listener, app).await?; + Ok(()) +} diff --git a/hive-subagent-mcp/src/paths.rs b/hive-subagent-mcp/src/paths.rs new file mode 100644 index 00000000..8a05e334 --- /dev/null +++ b/hive-subagent-mcp/src/paths.rs @@ -0,0 +1,15 @@ +//! The one per-agent path this daemon needs. + +use std::path::PathBuf; + +/// The harness-served in-agent todo socket (loose-ends v2). This daemon +/// pushes exactly one todo per subagent lifetime — the completion summary, +/// once its background turn finishes. Override via `HIVE_AGENT_SOCKET`; +/// same resolution as `hive-bash-mcp`'s own `paths::agent_socket`. +#[must_use] +pub fn agent_socket() -> PathBuf { + std::env::var_os("HIVE_AGENT_SOCKET").map_or_else( + || PathBuf::from(hive_agent_sock::DEFAULT_AGENT_SOCKET), + PathBuf::from, + ) +} diff --git a/hive-subagent-mcp/src/session.rs b/hive-subagent-mcp/src/session.rs new file mode 100644 index 00000000..d2a503d6 --- /dev/null +++ b/hive-subagent-mcp/src/session.rs @@ -0,0 +1,317 @@ +//! The claude-facing half of this daemon: spawn a subagent turn, track it +//! only while it's alive, and push exactly one todo when it finishes. +//! +//! **No task files, no restart recovery.** The daemon's only state is an +//! in-memory `name -> Cancel` map, live for exactly as long as the process +//! is — a daemon restart means whatever was running gets killed with it +//! (`tokio`'s own child-process drop semantics), not adopted. The durable +//! record of a subagent's existence is `hive_claude::SessionStore` — claude's +//! own on-disk session, found again by name. `continue` is how a caller +//! reattaches to it, whether that's "give it a new turn" or "the daemon +//! restarted and I want to pick this back up." +//! +//! **No mid-turn compaction.** Building on `hive_claude::Claude::spawn` + +//! `RunningClaude::wait` directly (not `InfiniteSession::run`) is what makes +//! `interrupt` possible at all — `InfiniteSession` has no cancel handle to +//! reach in from the outside, only `RunningClaude::cancel_handle` does. The +//! trade: this daemon doesn't get `InfiniteSession`'s reactive-compact-on- +//! overflow or proactive-checkpoint-compact for free: a turn that overflows +//! the context window surfaces as a plain `Error::PromptTooLong` to the +//! caller instead of self-healing. Subagents are meant to be bounded, +//! single-batch work (see the `base:claude-subagents` skill), not sessions +//! long-lived enough to need in-place compaction — a real follow-up if that +//! assumption stops holding, not shipped here. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::{Arc, Mutex, PoisonError}; + +use hive_claude::{Attach, Cancel, Claude, Config, NoopSink, SessionStore}; + +/// This daemon's whole state: which names currently have a live process, +/// and where to push the completion todo. `Arc`-wrapped so the background +/// task that drives a turn to completion can outlive the tool call that +/// started it. +pub struct State { + running: Mutex>, + socket: PathBuf, +} + +impl State { + #[must_use] + pub fn new(socket: PathBuf) -> Self { + Self { + running: Mutex::new(HashMap::new()), + socket, + } + } + + fn is_running(&self, name: &str) -> bool { + self.running + .lock() + .unwrap_or_else(PoisonError::into_inner) + .contains_key(name) + } +} + +/// A caller-chosen name, validated the same way `hive-bash-mcp`'s task ids +/// are: a single safe [`hive_types::Ident`] segment, which doubles as +/// claude's own `--name`/`--resume` session title. +fn validate_name(name: &str) -> anyhow::Result<()> { + hive_types::Ident::parse(name) + .map(|_| ()) + .map_err(|e| anyhow::anyhow!("invalid subagent name {name:?}: {e}")) +} + +/// Extend the ambient `OTEL_RESOURCE_ATTRIBUTES` with a `subagent=` +/// attribute, so every token/cost/tool-call data point this subagent's own +/// claude process emits carries it alongside the parent's `agent=` +/// label. `Config.env` applies after the inherited environment, so this one +/// entry overriding the ambient value is the intended shape, not a +/// wholesale replacement. +fn subagent_otel_attrs(name: &str) -> String { + match std::env::var("OTEL_RESOURCE_ATTRIBUTES") { + Ok(existing) if !existing.is_empty() => format!("{existing},subagent={name}"), + _ => format!("subagent={name}"), + } +} + +/// Build the `Config` one subagent turn runs against. `prompt_file`, when +/// given, becomes `--append-system-prompt-file` — the subagent's task +/// instructions. Always `--dangerously-skip-permissions --strict-mcp-config` +/// (no `--mcp-config` override — a safety property, not a knob). +fn build_config(name: &str, model: Option, prompt_file: Option<&str>) -> Config { + let mut extra_args = vec!["--dangerously-skip-permissions".to_owned()]; + if let Some(path) = prompt_file { + extra_args.push("--append-system-prompt-file".to_owned()); + extra_args.push(path.to_owned()); + } + Config { + model, + strict_mcp_config: true, + extra_args, + env: vec![( + "OTEL_RESOURCE_ATTRIBUTES".to_owned(), + subagent_otel_attrs(name), + )], + ..Default::default() + } +} + +/// The [`SessionStore`] a subagent's turn actually runs against — same +/// resolution `hive_claude::Claude` itself uses, so a lookup here can't +/// disagree with what the driver does a moment later. +fn build_store(config: &Config) -> std::io::Result { + Ok(SessionStore::new( + config.resolved_claude_home()?, + config.resolved_cwd()?, + )) +} + +/// Start a fresh subagent under `name`. A prior *finished* session under +/// the same name is archived first (so this is a real fresh start, not a +/// silent resume of old history) — a *currently running* one is refused +/// outright, since `hive_claude::InfiniteSession`'s own docs warn that two +/// concurrent runs against the same name corrupt both. +/// +/// # Errors +/// +/// A name already running, an invalid name, an archive failure, or the +/// underlying `Claude::spawn` failing (binary missing, etc.) — the last +/// case is the only one that can happen *after* commit-to-run, and it's +/// exactly why nothing is registered in `running` until spawn actually +/// succeeds. +pub fn start( + state: &Arc, + name: &str, + model: Option, + prompt_file: &str, + trigger: String, +) -> anyhow::Result { + validate_name(name)?; + if state.is_running(name) { + anyhow::bail!("subagent `{name}` is already running — use `continue` or `interrupt`"); + } + let config = build_config(name, model, Some(prompt_file)); + let store = build_store(&config)?; + if store.find_by_title(name).is_some() { + tracing::info!( + name, + "start: archiving a finished prior session for a fresh start" + ); + store + .archive_by_title(name) + .map_err(|e| anyhow::anyhow!("archiving the prior `{name}` session failed: {e}"))?; + } + spawn_and_track( + state, + name, + &config, + &Attach::Create(name.to_owned()), + trigger, + ) +} + +/// Give an existing named session a new turn — resuming it whether that +/// means "the previous turn finished, here's the next instruction" or "the +/// daemon restarted, reattaching." Refuses a name with no session on disk +/// at all (nothing to continue) or one already running (same +/// concurrent-run hazard as `start`). +/// +/// # Errors +/// +/// No session under `name`, an invalid name, one already running, or +/// `Claude::spawn` failing. +pub fn continue_( + state: &Arc, + name: &str, + prompt: String, + model: Option, +) -> anyhow::Result { + validate_name(name)?; + if state.is_running(name) { + anyhow::bail!( + "subagent `{name}` is already running — use `interrupt` first if you meant to redirect it" + ); + } + let config = build_config(name, model, None); + let store = build_store(&config)?; + if store.find_by_title(name).is_none() { + anyhow::bail!( + "no session named `{name}` exists — `continue` resumes an existing subagent, `start` \ + creates one" + ); + } + spawn_and_track( + state, + name, + &config, + &Attach::Resume(name.to_owned()), + prompt, + ) +} + +/// Spawn the child (synchronous — returns with a real pid the instant the +/// process exists, which *is* "confirmed running": there is no stronger +/// signal to wait for without slowing every call down for no reason), track +/// it in `running`, and hand the actual turn off to a background task so +/// the caller returns immediately instead of blocking on the whole turn. +/// Not `async` itself — `tokio::spawn` needs an active runtime to spawn +/// *onto*, not an `async` caller to spawn *from*. +fn spawn_and_track( + state: &Arc, + name: &str, + config: &Config, + attach: &Attach, + prompt: String, +) -> anyhow::Result { + let running = Claude::spawn(config, attach) + .map_err(|e| anyhow::anyhow!("starting the subagent process failed: {e}"))?; + state + .running + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert(name.to_owned(), running.cancel_handle()); + + let state = Arc::clone(state); + let task_name = name.to_owned(); + tokio::spawn(async move { + let outcome = running.wait(&prompt, &NoopSink).await; + state + .running + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(&task_name); + let summary = match outcome { + Ok(()) => "turn complete".to_owned(), + Err(e) => { + tracing::warn!(name = %task_name, error = %e, "subagent: turn failed"); + format!("claude error: {e}") + } + }; + push_completion_todo(&state.socket, &task_name, &summary).await; + }); + + Ok(format!("subagent `{name}` started")) +} + +/// Signal `name`'s running process — `force` picks SIGKILL over SIGINT (see +/// `hive_claude::Cancel::cancel`). Refuses a name with nothing running: +/// there's no queued/pending state to cancel pre-emptively any more (see +/// the module doc), only "running" or "not tracked." +/// +/// # Errors +/// +/// An invalid name, or nothing currently running under `name`. +pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result { + validate_name(name)?; + let mut running = state.running.lock().unwrap_or_else(PoisonError::into_inner); + let Some(cancel) = running.remove(name) else { + anyhow::bail!("no subagent named `{name}` is currently running"); + }; + cancel.cancel(force); + Ok(format!("interrupt sent to subagent `{name}`")) +} + +/// Push `name`'s one-shot completion todo. Best-effort: a connect/write +/// failure is logged and swallowed, matching every other in-agent-socket +/// producer in this codebase — there's no retry queue to fall back to, and +/// the caller has already moved on by the time this fires. +async fn push_completion_todo(socket: &std::path::Path, name: &str, summary: &str) { + let req = hive_agent_sock::Request::UpsertTodo { + subsystem: "subagent".to_owned(), + key: Some(name.to_owned()), + summary: format!("subagent `{name}` finished: {summary}"), + source: None, + reopen_if_acked: false, + }; + if let Err(e) = hive_sock_client::notify(socket, &req, hive_sock_client::Retry::None).await { + tracing::warn!(name, error = ?e, "subagent: completion todo push failed"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn otel_attrs_appends_when_ambient_var_is_set() { + // SAFETY: test-only env mutation, single-threaded within this fn's + // scope (no other test in this crate touches this var — checked). + unsafe { + std::env::set_var("OTEL_RESOURCE_ATTRIBUTES", "agent=damocles"); + } + assert_eq!( + subagent_otel_attrs("batch-1"), + "agent=damocles,subagent=batch-1" + ); + unsafe { + std::env::remove_var("OTEL_RESOURCE_ATTRIBUTES"); + } + } + + #[test] + fn otel_attrs_stands_alone_when_ambient_var_is_unset() { + unsafe { + std::env::remove_var("OTEL_RESOURCE_ATTRIBUTES"); + } + assert_eq!(subagent_otel_attrs("batch-1"), "subagent=batch-1"); + } + + #[test] + fn build_config_only_appends_system_prompt_when_given() { + let with = build_config("n", None, Some("/tmp/p.md")); + assert!( + with.extra_args + .contains(&"--append-system-prompt-file".to_owned()) + ); + assert!(with.extra_args.contains(&"/tmp/p.md".to_owned())); + + let without = build_config("n", None, None); + assert!( + !without + .extra_args + .contains(&"--append-system-prompt-file".to_owned()) + ); + } +}