//! 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), 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 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`.
pub struct State {
    running: Mutex<HashMap<String, Option<Cancel>>>,
    socket: PathBuf,
}

impl State {
    #[must_use]
    pub fn new(socket: PathBuf) -> Self {
        Self {
            running: 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);
        }
    }
}

/// 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.reserve(name) {
        anyhow::bail!("subagent `{name}` is already running — use `continue` or `interrupt`");
    }
    let result = start_reserved(state, name, model, prompt_file, trigger);
    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,
) -> anyhow::Result<String> {
    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.reserve(name) {
        anyhow::bail!(
            "subagent `{name}` is already running — use `interrupt` first if you meant to redirect it"
        );
    }
    let result = continue_reserved(state, name, prompt, model);
    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>,
) -> anyhow::Result<String> {
    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}"))?;
    // 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) -> anyhow::Result<String> {
    validate_name(name)?;
    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);
    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"
        );
    }

    #[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())
        );
    }
}