675 lines
27 KiB
Rust
675 lines
27 KiB
Rust
//! 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 -> Option<Cancel>` map (see `State`'s own doc for what
|
|
//! the `None`/`Some` split is for) plus a `name -> dir` memory so `start`'s
|
|
//! `dir` doesn't have to be repeated on every later `continue`/`status`
|
|
//! (`State::resolve_dir`) — both 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, and the
|
|
//! remembered `dir` is gone too. 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" — after a restart, re-supply
|
|
//! `dir` once if the session isn't in the daemon's own default directory.
|
|
//!
|
|
//! **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 or
|
|
/// a reservation in flight, 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.
|
|
///
|
|
/// The map value is `Option<Cancel>`: `None` means `name` is reserved for
|
|
/// an in-flight `start`/`continue` that hasn't reached a confirmed
|
|
/// `Claude::spawn` yet; `Some(cancel)` means a real process is tracked and
|
|
/// interruptible. The `None` state exists to close a real TOCTOU window a
|
|
/// reviewer caught in the original check-then-insert version: checking "is
|
|
/// `name` free" and committing to it are two different lock acquisitions
|
|
/// unless the check *is* the reservation — see `reserve`.
|
|
///
|
|
/// ⚠️ This concurrency guard is keyed by `name` alone — a `start`/`continue`
|
|
/// for `name` with a *different* `dir` than one already in flight under
|
|
/// that name is refused as "already running," even though the two would
|
|
/// resolve to entirely separate on-disk sessions. Deliberate: `name` is the
|
|
/// caller's one chosen identity for a subagent, not `(name, dir)` — reuse a
|
|
/// name across directories at your own risk, the tool doesn't disambiguate
|
|
/// it (flagged in review when `dir` was added).
|
|
pub struct State {
|
|
running: Mutex<HashMap<String, Option<Cancel>>>,
|
|
dirs: Mutex<HashMap<String, String>>,
|
|
socket: PathBuf,
|
|
}
|
|
|
|
impl State {
|
|
#[must_use]
|
|
pub fn new(socket: PathBuf) -> Self {
|
|
Self {
|
|
running: Mutex::new(HashMap::new()),
|
|
dirs: Mutex::new(HashMap::new()),
|
|
socket,
|
|
}
|
|
}
|
|
|
|
/// `Some(true)` — a live process is tracked, interruptible. `Some(false)`
|
|
/// — reserved for an in-flight start/continue, not yet a confirmed
|
|
/// spawn. `None` — nothing tracked under `name` at all.
|
|
fn occupancy(&self, name: &str) -> Option<bool> {
|
|
self.running
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner)
|
|
.get(name)
|
|
.map(Option::is_some)
|
|
}
|
|
|
|
/// Atomically claim `name` for an in-flight start/continue: check
|
|
/// "is anything tracked under this name" and "commit to this call
|
|
/// owning it" in the *same* lock acquisition, so two calls racing the
|
|
/// same name can't both pass a check before either commits (the exact
|
|
/// same-name concurrent-run hazard this module's doc warns about).
|
|
/// Returns `false` (reserving nothing) if `name` is already reserved
|
|
/// or running.
|
|
fn reserve(&self, name: &str) -> bool {
|
|
let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner);
|
|
if running.contains_key(name) {
|
|
return false;
|
|
}
|
|
running.insert(name.to_owned(), None);
|
|
true
|
|
}
|
|
|
|
/// Release a reservation that never made it to a real spawn (an error
|
|
/// on the slow path between `reserve` and `Claude::spawn` succeeding).
|
|
/// A no-op if the entry was already upgraded to `Some` — this only ever
|
|
/// clears a still-`None` placeholder, never a live process.
|
|
fn release_reservation(&self, name: &str) {
|
|
let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner);
|
|
if matches!(running.get(name), Some(None)) {
|
|
running.remove(name);
|
|
}
|
|
}
|
|
|
|
/// An explicit `dir` is remembered against `name` and returned as-is; an
|
|
/// omitted one (`None`) falls back to whatever was last remembered for
|
|
/// `name`, so a `start` that gave a `dir` doesn't force every later
|
|
/// `continue`/`status` against the same name to repeat it. Explicit
|
|
/// always wins and updates the memory — there's no way to say "forget
|
|
/// it, use the default" once something's been remembered short of
|
|
/// giving a genuinely different `dir`, a deliberate simplicity trade,
|
|
/// not an oversight. In-memory only, same durability envelope as
|
|
/// `running`: a daemon restart forgets it same as it forgets everything
|
|
/// else here — the caller re-supplies `dir` once, same as it always
|
|
/// could.
|
|
///
|
|
/// ⚠️ **Callers must only call this once the call is known to proceed**
|
|
/// (i.e. after `reserve` has already succeeded in `start`/`continue_`) —
|
|
/// see `peek_dir` for the non-committing variant a call that might still
|
|
/// be refused (or that shouldn't persist its `dir` at all, like
|
|
/// `status`) needs instead. This one argus caught in review: calling it
|
|
/// unconditionally before `reserve`'s check meant a *rejected* concurrent
|
|
/// call still overwrote the remembered `dir` for the name it was refused
|
|
/// against.
|
|
fn resolve_dir(&self, name: &str, dir: Option<&str>) -> Option<String> {
|
|
let mut dirs = self.dirs.lock().unwrap_or_else(PoisonError::into_inner);
|
|
match dir {
|
|
Some(d) => {
|
|
dirs.insert(name.to_owned(), d.to_owned());
|
|
Some(d.to_owned())
|
|
}
|
|
None => dirs.get(name).cloned(),
|
|
}
|
|
}
|
|
|
|
/// Same resolution as `resolve_dir` (explicit wins, omitted falls back
|
|
/// to the remembered value) but never writes — for a call that must not
|
|
/// change what a later bare `continue`/`status` resolves to. `status`'s
|
|
/// own doc offers `dir` as a one-off "check a different directory"
|
|
/// knob; if that call committed the same way `start`/`continue_` do, the
|
|
/// one-off peek would silently become the new remembered default (the
|
|
/// second bug argus flagged in the same review).
|
|
fn peek_dir(&self, name: &str, dir: Option<&str>) -> Option<String> {
|
|
match dir {
|
|
Some(d) => Some(d.to_owned()),
|
|
None => self
|
|
.dirs
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner)
|
|
.get(name)
|
|
.cloned(),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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. `dir`, when given, becomes `Config::cwd` (e.g. a worktree
|
|
/// the caller already prepared); `None` inherits this daemon's own working
|
|
/// directory, same as before this field existed. 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>,
|
|
dir: 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,
|
|
cwd: dir.map(PathBuf::from),
|
|
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,
|
|
dir: Option<&str>,
|
|
) -> anyhow::Result<String> {
|
|
validate_name(name)?;
|
|
if !state.reserve(name) {
|
|
anyhow::bail!("subagent `{name}` is already running — use `continue` or `interrupt`");
|
|
}
|
|
// Only commit the remembered `dir` now that `reserve` has actually
|
|
// claimed `name` — see `resolve_dir`'s doc for why the order matters.
|
|
let dir = state.resolve_dir(name, dir);
|
|
let result = start_reserved(state, name, model, prompt_file, trigger, dir.as_deref());
|
|
if result.is_err() {
|
|
state.release_reservation(name);
|
|
}
|
|
result
|
|
}
|
|
|
|
/// The slow, fallible part of `start`, run only after `reserve` has
|
|
/// already closed the TOCTOU window — split out so `start` can release the
|
|
/// reservation on any error path here without duplicating that logic per
|
|
/// failure site.
|
|
fn start_reserved(
|
|
state: &Arc<State>,
|
|
name: &str,
|
|
model: Option<String>,
|
|
prompt_file: &str,
|
|
trigger: String,
|
|
dir: Option<&str>,
|
|
) -> anyhow::Result<String> {
|
|
let config = build_config(name, model, Some(prompt_file), dir);
|
|
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>,
|
|
dir: Option<&str>,
|
|
) -> anyhow::Result<String> {
|
|
validate_name(name)?;
|
|
if !state.reserve(name) {
|
|
anyhow::bail!(
|
|
"subagent `{name}` is already running — use `interrupt` first if you meant to redirect it"
|
|
);
|
|
}
|
|
// Only commit the remembered `dir` now that `reserve` has actually
|
|
// claimed `name` — see `resolve_dir`'s doc for why the order matters.
|
|
let dir = state.resolve_dir(name, dir);
|
|
let result = continue_reserved(state, name, prompt, model, dir.as_deref());
|
|
if result.is_err() {
|
|
state.release_reservation(name);
|
|
}
|
|
result
|
|
}
|
|
|
|
/// The slow, fallible part of `continue_`, run only after `reserve` has
|
|
/// already closed the TOCTOU window — same split rationale as
|
|
/// `start_reserved`.
|
|
fn continue_reserved(
|
|
state: &Arc<State>,
|
|
name: &str,
|
|
prompt: String,
|
|
model: Option<String>,
|
|
dir: Option<&str>,
|
|
) -> anyhow::Result<String> {
|
|
let config = build_config(name, model, None, dir);
|
|
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}"))?;
|
|
// Upgrades the `None` reservation `reserve` already placed here to a
|
|
// real cancel handle — same key, so there's no window where `name`
|
|
// reads as unoccupied between the reservation and this insert.
|
|
state
|
|
.running
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner)
|
|
.insert(name.to_owned(), Some(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"))
|
|
}
|
|
|
|
/// Report whether `name` is currently running — a zero-cost check that
|
|
/// never launches a process, unlike `continue`. Distinguishes four states:
|
|
/// running, starting (reserved, not yet a confirmed spawn — see `State`'s
|
|
/// doc), idle (a session exists but nothing is in flight), and no such
|
|
/// session at all.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// An invalid name, or no session — running or on disk — under `name`.
|
|
pub fn status(state: &State, name: &str, dir: Option<&str>) -> anyhow::Result<String> {
|
|
validate_name(name)?;
|
|
// Read-only: an explicit `dir` here is a one-off "check this other
|
|
// directory's session" per this fn's own doc, not a new remembered
|
|
// default — `peek_dir` resolves the same way but never writes.
|
|
let dir = state.peek_dir(name, dir);
|
|
match state.occupancy(name) {
|
|
Some(true) => return Ok(format!("subagent `{name}` is running")),
|
|
Some(false) => {
|
|
return Ok(format!(
|
|
"subagent `{name}` is starting — not yet confirmed running"
|
|
));
|
|
}
|
|
None => {}
|
|
}
|
|
let config = build_config(name, None, None, dir.as_deref());
|
|
let store = build_store(&config)?;
|
|
if store.find_by_title(name).is_some() {
|
|
Ok(format!(
|
|
"subagent `{name}` is idle — its last turn finished; `continue` to give it another"
|
|
))
|
|
} else {
|
|
anyhow::bail!("no subagent named `{name}` exists — `start` creates one")
|
|
}
|
|
}
|
|
|
|
/// Signal `name`'s running process — `force` picks SIGKILL over SIGINT (see
|
|
/// `hive_claude::Cancel::cancel`). Refuses a name with nothing running: no
|
|
/// entry at all, or one still in the brief `reserve`d-but-not-yet-spawned
|
|
/// window (nothing to signal yet — the reservation is put back so a
|
|
/// concurrent `start`/`continue` for the same name still gets refused).
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// An invalid name, nothing tracked under `name`, or `name` is still
|
|
/// starting (reserved, not yet a confirmed spawn).
|
|
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);
|
|
match running.remove(name) {
|
|
None => anyhow::bail!("no subagent named `{name}` is currently running"),
|
|
Some(None) => {
|
|
running.insert(name.to_owned(), None);
|
|
anyhow::bail!(
|
|
"subagent `{name}` is still starting — not yet confirmed running, try again \
|
|
shortly"
|
|
);
|
|
}
|
|
Some(Some(cancel)) => {
|
|
drop(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::*;
|
|
|
|
// `Cancel` can only be constructed from a real spawned process (no test
|
|
// fixture in `hive_claude` for it), so these exercise the reservation
|
|
// half of `State` directly — the actual TOCTOU-closure logic — rather
|
|
// than the full start/spawn path.
|
|
|
|
#[test]
|
|
fn reserve_is_exclusive_for_the_same_name() {
|
|
let state = State::new(PathBuf::from("/dev/null"));
|
|
assert!(state.reserve("dup"), "first reservation should succeed");
|
|
assert!(
|
|
!state.reserve("dup"),
|
|
"a second reservation for the same name must be refused — this is the exact race \
|
|
argus found: two calls both passing a check before either commits"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn reserve_does_not_cross_block_different_names() {
|
|
let state = State::new(PathBuf::from("/dev/null"));
|
|
assert!(state.reserve("a"));
|
|
assert!(
|
|
state.reserve("b"),
|
|
"unrelated names must not block each other"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn release_reservation_frees_the_name_for_reuse() {
|
|
let state = State::new(PathBuf::from("/dev/null"));
|
|
assert!(state.reserve("n"));
|
|
state.release_reservation("n");
|
|
assert!(
|
|
state.reserve("n"),
|
|
"releasing a still-`None` reservation must free the name again"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn occupancy_reflects_the_reserved_but_not_running_state() {
|
|
let state = State::new(PathBuf::from("/dev/null"));
|
|
assert_eq!(state.occupancy("never-reserved"), None);
|
|
state.reserve("n");
|
|
assert_eq!(
|
|
state.occupancy("n"),
|
|
Some(false),
|
|
"reserved-but-not-yet-spawned must read as occupied-but-not-running"
|
|
);
|
|
}
|
|
|
|
// One test, not two: `subagent_otel_attrs` reads the real process-wide
|
|
// `OTEL_RESOURCE_ATTRIBUTES` env var, and cargo runs tests in parallel
|
|
// threads by default — two separate tests each mutating that global
|
|
// raced each other (and, in this container, lost to the agent's own
|
|
// real ambient value). Sequencing both assertions in one test removes
|
|
// the race instead of papering over it with a mutex.
|
|
#[test]
|
|
fn otel_attrs_append_ambient_value_or_stand_alone() {
|
|
// SAFETY: test-only env mutation; sequenced within this one test so
|
|
// no other test's concurrent read/write of the same var can race it.
|
|
unsafe {
|
|
std::env::remove_var("OTEL_RESOURCE_ATTRIBUTES");
|
|
}
|
|
assert_eq!(subagent_otel_attrs("batch-1"), "subagent=batch-1");
|
|
|
|
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 build_config_only_appends_system_prompt_when_given() {
|
|
let with = build_config("n", None, Some("/tmp/p.md"), None);
|
|
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, None);
|
|
assert!(
|
|
!without
|
|
.extra_args
|
|
.contains(&"--append-system-prompt-file".to_owned())
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn resolve_dir_remembers_an_explicit_dir_and_falls_back_to_it_when_omitted() {
|
|
let state = State::new(PathBuf::from("/dev/null"));
|
|
assert_eq!(
|
|
state.resolve_dir("n", None),
|
|
None,
|
|
"nothing remembered yet — omitted dir has nothing to fall back to"
|
|
);
|
|
assert_eq!(
|
|
state.resolve_dir("n", Some("/tmp/worktree")),
|
|
Some("/tmp/worktree".to_owned()),
|
|
"an explicit dir is returned as-is"
|
|
);
|
|
assert_eq!(
|
|
state.resolve_dir("n", None),
|
|
Some("/tmp/worktree".to_owned()),
|
|
"a later omitted dir falls back to what was just remembered"
|
|
);
|
|
assert_eq!(
|
|
state.resolve_dir("n", Some("/tmp/other")),
|
|
Some("/tmp/other".to_owned()),
|
|
"a later explicit dir overrides the memory, not just reads it"
|
|
);
|
|
assert_eq!(
|
|
state.resolve_dir("n", None),
|
|
Some("/tmp/other".to_owned()),
|
|
"the fallback now reflects the override"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn resolve_dir_does_not_cross_names() {
|
|
let state = State::new(PathBuf::from("/dev/null"));
|
|
state.resolve_dir("a", Some("/tmp/a"));
|
|
assert_eq!(
|
|
state.resolve_dir("b", None),
|
|
None,
|
|
"an unrelated name's memory must stay empty"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn peek_dir_resolves_like_resolve_dir_but_never_writes() {
|
|
let state = State::new(PathBuf::from("/dev/null"));
|
|
state.resolve_dir("n", Some("/tmp/remembered"));
|
|
assert_eq!(
|
|
state.peek_dir("n", Some("/tmp/one-off")),
|
|
Some("/tmp/one-off".to_owned()),
|
|
"an explicit dir still resolves as given"
|
|
);
|
|
assert_eq!(
|
|
state.resolve_dir("n", None),
|
|
Some("/tmp/remembered".to_owned()),
|
|
"but the one-off peek must not have overwritten the remembered value"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn start_does_not_corrupt_the_remembered_dir_when_the_name_is_already_running() {
|
|
// The exact scenario argus caught in review: a rejected concurrent
|
|
// `start` used to commit its `dir` before the reservation check
|
|
// refused it, corrupting what a later bare `continue` would resolve
|
|
// to. Reordering `reserve` before `resolve_dir` in `start` closes it
|
|
// — a name that's already reserved must never reach `resolve_dir` at
|
|
// all, so `dirs` stays exactly as a caller left it.
|
|
let state = Arc::new(State::new(PathBuf::from("/dev/null")));
|
|
state.resolve_dir("dup", Some("/tmp/original"));
|
|
assert!(
|
|
state.reserve("dup"),
|
|
"simulate an in-flight start/continue already owning the name"
|
|
);
|
|
let result = start(
|
|
&state,
|
|
"dup",
|
|
None,
|
|
"/tmp/prompt.md",
|
|
"trigger".to_owned(),
|
|
Some("/tmp/rejected"),
|
|
);
|
|
assert!(
|
|
result.is_err(),
|
|
"a name already reserved must be refused, not spawned"
|
|
);
|
|
assert_eq!(
|
|
state.resolve_dir("dup", None),
|
|
Some("/tmp/original".to_owned()),
|
|
"the rejected call's dir must not have overwritten the remembered one"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn build_config_sets_cwd_only_when_a_dir_is_given() {
|
|
let with = build_config("n", None, None, Some("/tmp/some-worktree"));
|
|
assert_eq!(with.cwd, Some(PathBuf::from("/tmp/some-worktree")));
|
|
|
|
let without = build_config("n", None, None, None);
|
|
assert_eq!(without.cwd, None);
|
|
}
|
|
}
|