hive-subagent-mcp: new crate for the subagent daemon, independent of hive-bash-mcp

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<HashMap<String, Cancel>>,
+    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=<name>`
+/// attribute, so every token/cost/tool-call data point this subagent's own
+/// claude process emits carries it alongside the parent's `agent=<name>`
+/// 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<String>, 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<SessionStore> {
+    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<State>,
+    name: &str,
+    model: Option<String>,
+    prompt_file: &str,
+    trigger: String,
+) -> anyhow::Result<String> {
+    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<State>,
+    name: &str,
+    prompt: String,
+    model: Option<String>,
+) -> anyhow::Result<String> {
+    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<State>,
+    name: &str,
+    config: &Config,
+    attach: &Attach,
+    prompt: String,
+) -> anyhow::Result<String> {
+    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<String> {
+    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())
+        );
+    }
+}