trusty-mpm 1.8.2

//! Non-opt-in delegation tracking from Claude Code hooks (#2864 S2).
//!
//! Why: delegation tracking had two defects that together made it useless as a
//! basis for "which agents are in flight?". (1) It was opt-in — a record existed
//! only if the PM chose to call the `agent_delegate` MCP tool, so a PM that
//! dispatched work with the native subagent tool alone left no trace. (2) Status
//! never moved: the only production construction is
//! [`Delegation::new`](crate::core::agent::Delegation::new), which yields
//! `Queued`, and nothing ever wrote `Running`/`Completed`/`Failed`. So
//! [`has_live_children`](crate::daemon::idle_nudge::has_live_children) — which
//! treats `Queued` and `Running` as live — reported *every* delegation as live
//! forever, permanently suppressing the idle nudge for any session that had ever
//! delegated once.
//!
//! This module fixes both by observing hooks the daemon already receives: a
//! `PreToolUse` with `matcher: "*"` is installed on every managed session, so
//! every native subagent dispatch already arrives here. No new hook, no opt-in.
//!
//! # Correlation model (empirically established, #2864 Step 0)
//!
//! Captured from live Claude Code 2.1.220 stdin with two *concurrent*
//! subagents — the case that breaks any "most recent delegation" guess:
//!
//! ```text
//! PreToolUse   tool_name=Agent  tool_use_id=toolu_01DA…   (no agent id yet)
//! PostToolUse  tool_name=Agent  tool_use_id=toolu_01DA…
//!              tool_response={ isAsync: true,
//!                              status: "async_launched",
//!                              agentId: "a403cdbc078b5c474" }
//! SubagentStop                  agent_id="a403cdbc078b5c474"
//! ```
//!
//! Two consequences drive the whole design, and **neither may be "simplified"
//! away without re-running the probe**:
//!
//! 1. **`PostToolUse` does NOT mean the subagent finished.** For an async
//!    dispatch it fires ~1 ms after launch (`duration_ms: 1`) with
//!    `status: "async_launched"`, while the subagent runs on for minutes.
//!    Terminalizing there would mark every delegation `Completed` almost
//!    immediately and report "no agents in flight" while they are all still
//!    running — strictly worse than not tracking at all. `PostToolUse` is
//!    therefore a *join* step (it teaches us `agentId`), and terminalizes only
//!    when the dispatch genuinely returned synchronously.
//! 2. **`SubagentStop` DOES carry an exact correlation key** — `agent_id`,
//!    identical to `tool_response.agentId`. It is the authoritative terminal
//!    signal, not an advisory one. Because the key is exact, closing one of two
//!    concurrent delegations cannot close the other.
//!
//! Absent an `agent_id` (an older Claude Code, or a dispatch whose `PostToolUse`
//! was missed) a `SubagentStop` terminalizes **nothing**. Guessing "the most
//! recent Running delegation" would close the wrong one under concurrency and
//! manufacture a false idle, which is the specific failure this design exists to
//! prevent.
//!
//! # Arrival order is a race, not a contract (#4142)
//!
//! `PostToolUse` is installed `async: true` and `SubagentStop` synchronously
//! (`core::standalone::hooks`), so each is its own `tm hook` process posting
//! independently. The async one can be descheduled while the synchronous stop
//! for the same subagent completes and posts first — at which point the stop
//! quotes an `agent_id` no record has learned yet, and the only reason it
//! resolves nothing is that it is early.
//!
//! [`on_subagent_stop`] therefore *holds* an unresolvable stop in
//! [`crate::daemon::state::pending_stops`] rather than discarding it, and
//! [`on_launched`] applies it the instant the `agent_id` is taught. The key is
//! the exact id the stop quoted, so the deferred path terminalizes precisely the
//! record the in-order path would have — the two orders converge instead of one
//! of them winning. This buffers evidence; it never invents any, and the "most
//! recent" guess stays forbidden in both directions.
//!
//! # Fail direction
//!
//! Every branch here fails **closed**: uncertainty leaves a delegation `Running`
//! and never terminalizes it. A false "still running" delays an idle nudge; a
//! false "completed" tells a human that no agents are in flight when several
//! are, which is the one report this feature must never produce. So
//! [`on_subagent_stop`] acts only on an exact `agent_id`, and [`on_launched`]
//! acts only on a `tool_response` it [`recognized_response`]-ised — an absent or
//! unrecognized response is "we do not know", not "it finished".
//!
//! The cost of failing closed is that a delegation can enter `Running` with no
//! route out: `SubagentStop` is not guaranteed to arrive (`tm hook` POSTs fail
//! open on a 2 s budget, an interrupted subagent emits no stop, a dispatch that
//! never learned an `agent_id` can never be resolved by one — a stop that merely
//! arrived early is the one case now recovered, see #4142 above). That is bounded
//! outside the hot path by
//! [`DaemonState::sweep_delegations`](crate::daemon::state::DaemonState::sweep_delegations),
//! which the 60 s reap loop drives: a long-overdue live delegation becomes
//! `Stale` — explicitly *not* `Completed`, and still recoverable by a late stop —
//! so it stops suppressing the idle nudge without anyone ever being told it
//! finished.
//!
//! # Cost
//!
//! [`observe`] runs inside the synchronous `PreToolUse` hook, which has a 5 s
//! timeout and sits in front of *every* tool call in *every* managed session.
//! The non-delegation path — which is essentially all traffic — is one
//! `payload.get("tool")` plus an exact string compare against two constants, and
//! then returns. The delegation path is in-memory `DashMap` work only: no disk,
//! no network, no async, no blocking lock.
//!
//! # Known limitation
//!
//! Records are keyed by the session id the event arrives under: the Claude
//! session UUID for hook-observed records (matching
//! [`crate::daemon::idle_nudge::maybe_nudge_parked_session`]), but whatever id
//! the PM passed for `agent_delegate` records. When those differ, [`dedup`]
//! cannot match and a declaration plus its observation remain two records. That
//! id ambiguity predates #2864 and is not addressed here.
//!
//! Test: the `#[cfg(test)]` suite below.

use chrono::Utc;
use serde_json::Value;

use crate::core::agent::{
    Delegation, DelegationId, DelegationSource, DelegationStatus, ModelTier, TASK_STOP_TOOL,
    TOOL_RESPONSE_KEYS, is_subagent_dispatch_tool,
};
use crate::core::dispatch_isolation::{dispatch_isolation, isolation_separates_working_tree};
use crate::core::hook::HookEvent;
use crate::core::session::SessionId;
use crate::daemon::state::DaemonState;
use crate::session_manager::worktree_ownership::{
    AgentWorktreeOwner, SentinelOwner, read_sentinel_owner, write_agent_sentinel,
};

use crate::session_manager::worktree_ownership::is_harness_agent_worktree;

/// How long after an `agent_delegate` call an observed dispatch may be treated
/// as the *same* delegation.
///
/// Why: a PM that both declares (`agent_delegate`) and dispatches (native tool)
/// must produce one record, not two. The declaration immediately precedes the
/// dispatch *in the same turn*, so the true gap is seconds; the window only has
/// to absorb a slow turn, not a whole work item. It was 300 s, which is long
/// enough for a PM that dispatches the same agent repeatedly — this repo
/// routinely runs several `rust-engineer`s — to have an unrelated later dispatch
/// absorbed into an earlier declaration (#2864 review, MEDIUM 2). Two minutes
/// keeps the legitimate merge and cuts the collision surface; [`tasks_match`] is
/// the second, independent discriminator.
/// Test: `dedups_declaration_and_observation`, `dedup_window_expires`.
const DEDUP_WINDOW_SECS: i64 = 120;

/// Observe one hook event and update delegation tracking.
///
/// Why: the single entry point [`crate::daemon::services::hook_service::HookService::process`]
/// calls, so the hook pipeline gains delegation tracking by way of one line and
/// all the correlation rules stay here.
/// What: routes a subagent-dispatch `PreToolUse` to [`on_dispatch`], its
/// `PostToolUse`/`PostToolUseFailure` to [`on_launched`], a `PostToolUse` on
/// any OTHER tool to [`on_task_stop`] (which acts only on `TaskStop`, #7487),
/// and `SubagentStop`/`SubagentStopFailure` to [`on_subagent_stop`]. Every other
/// event returns immediately. Never panics and never blocks; a payload missing
/// the fields it needs is skipped silently (fail-open — tracking is
/// observational and must never affect the hook verdict).
/// Test: `ignores_unrelated_tools`, `pre_tool_use_creates_running_delegation`,
/// and the rest of the suite below.
pub fn observe(state: &DaemonState, session: SessionId, event: HookEvent, payload: &Value) {
    match event {
        // Hot path: the overwhelming majority of PreToolUse events are ordinary
        // tools. Bail on the tool-name compare before touching any state.
        HookEvent::PreToolUse => {
            // #4311: a SUBAGENT's own tool call is the only event that names the
            // tree the harness gave it. Costs one `payload.get("agent_id")` on
            // the hot path — the same class as the tool-name compare below — and
            // writes only when the answer changed.
            register_agent_worktree(state, session, payload);
            if dispatch_tool(payload).is_some() {
                on_dispatch(state, session, payload);
            }
        }
        HookEvent::PostToolUse | HookEvent::PostToolUseFailure => {
            if dispatch_tool(payload).is_some() {
                on_launched(state, session, payload, event);
            } else if event == HookEvent::PostToolUse {
                // #7487: a stopped agent emits no `SubagentStop`, so this is the
                // only signal that its claim on a shared tree is over.
                on_task_stop(state, session, payload);
            }
        }
        HookEvent::SubagentStop | HookEvent::SubagentStopFailure => {
            on_subagent_stop(state, session, payload, event);
        }
        _ => {}
    }
}

/// Does this `tool_response` say the call it answers FAILED (#7487)?
///
/// Why: `TaskStop` "returns a success or failure status", and a stop that did
/// not take leaves the agent running in the tree. The wire shape is
/// `is_error` — the one [`TOOL_RESPONSE_KEYS`] member whose meaning is the same
/// for every tool, and the same key [`on_launched`] already trusts for exactly
/// this question. `HookEvent::PostToolUseFailure` is NOT the production signal:
/// no hook block `tm` writes registers that name, so it can never arrive.
/// What: `true` only for an explicit `is_error: true`. An absent, non-object or
/// `is_error`-less response is "cannot tell", which does not refuse — see
/// [`on_task_stop`]'s own note on why that is the right direction here.
/// Test: `a_failed_task_stop_releases_nothing`,
/// `a_task_stop_with_no_response_still_releases`.
fn stop_reported_failure(payload: &Value) -> bool {
    payload
        .get("tool_response")
        .and_then(|r| r.get("is_error"))
        .and_then(Value::as_bool)
        .unwrap_or(false)
}

/// `TaskStop` returned: the agent it names is gone, so release its claim.
///
/// Why: the tracker terminalizes on `SubagentStop`, and a subagent cancelled
/// with `TaskStop` emits none — it is killed, it does not finish. Its record
/// therefore stayed live for the six hours of `RUNNING_STALE_AFTER_SECS`, and
/// [`crate::daemon::state::DaemonState::shared_tree_occupants`] kept naming it,
/// so every later unisolated file-mutating dispatch into that directory was
/// denied for a collision that had already ended. Observed 2026-09-11: one
/// stopped `qa` agent blocked the cwd for the rest of the session (#7487).
/// What: on the `PostToolUse` for a successful `TaskStop`, terminalizes the
/// delegation whose `agent_id` equals the stopped `task_id` with
/// [`DelegationStatus::Cancelled`] — the status that already exists for exactly
/// this fact, and is terminal, so nothing counts the record as a writer again.
/// Returns whether it wrote.
///
/// Three refusals:
/// * [`stop_reported_failure`] — the stop did not take, so the agent is still
///   in the tree and releasing it would admit a second writer.
/// * No `task_id` (nor its deprecated `shell_id` alias) matches nothing.
/// * A `task_id` naming a background shell, a teammate, or an agent this
///   session never dispatched matches no record and writes nothing. The match
///   is on `agent_id` alone — the exact key `SubagentStop` uses — because a
///   "most recent live record" guess would close the wrong one under
///   concurrency, which is what the module note forbids. A dispatch whose
///   `agent_id` was never taught is unreachable here for the same reason it is
///   unreachable by a stop; the staleness sweep remains its only route out.
///
/// # A response that says nothing releases, and that is deliberate
///
/// This is the module's ONE fail-open branch, chosen against its own default
/// because the alternative is worse HERE. `TaskStop`'s response shape is not
/// pinned by any contract this repo owns, and `tm hook`'s projection keeps only
/// [`TOOL_RESPONSE_KEYS`] — so a plain-string or `is_error`-less answer yields
/// no forwarded response at all. Requiring positive proof of success would then
/// release nothing, ever, which is #7487 itself reinstated silently and
/// invisibly to every test. Refusing on the one shape that positively says
/// "failed" closes the hazard the response can actually express; the residual
/// is a stop that failed and said so in a spelling this cannot read, whose cost
/// is the pre-#7487 six-hour lock arriving by a different route.
///
/// Test: `a_task_stop_releases_the_stopped_agents_claim`,
/// `a_failed_task_stop_releases_nothing`,
/// `a_task_stop_with_no_response_still_releases`,
/// `a_task_stop_naming_an_unknown_id_terminalizes_nothing`.
fn on_task_stop(state: &DaemonState, session: SessionId, payload: &Value) -> bool {
    if payload.get("tool").and_then(Value::as_str) != Some(TASK_STOP_TOOL) {
        return false;
    }
    if stop_reported_failure(payload) {
        tracing::debug!("delegation: a TaskStop reported failure — the agent stays live (#7487)");
        return false;
    }
    let Some(task_id) = payload
        .get("input")
        .and_then(|i| i.get("task_id").or_else(|| i.get("shell_id")))
        .and_then(Value::as_str)
        .filter(|s| !s.is_empty())
    else {
        return false;
    };
    let Some(id) = state.find_delegation(session, |d| {
        d.agent_id.as_deref() == Some(task_id) && !d.status.is_terminal()
    }) else {
        return false;
    };
    tracing::info!(
        task_id,
        "delegation: a TaskStop cancelled this agent — releasing the tree it claimed (#7487)"
    );
    state.terminate_delegation(id, DelegationStatus::Cancelled)
}

/// The guard denied this dispatch: it never ran, so it holds nothing (#7487).
///
/// Why: two `PreToolUse` hooks see one dispatch. `tm hook --pm-guard` decides,
/// and the `matcher: "*"` tracker hook records — independently, and whatever
/// the guard decided. A denied dispatch therefore left a live record carrying
/// its `cwd`, which [`crate::daemon::state::DaemonState::shared_tree_occupants`]
/// then counted as a writer, so the deny for one collision manufactured the
/// next one. Observed 2026-09-11: the second refusal named "qa, rust-engineer",
/// the `rust-engineer` being the dispatch the guard had just refused (#7487).
/// What: with the dispatch-record lock held, terminalizes the record carrying
/// this `tool_use_id` as [`DelegationStatus::Cancelled`], or writes one in that
/// status when none exists yet. Returns whether it wrote.
///
/// The write-when-absent arm is not bookkeeping for its own sake: the two hooks
/// race, so the tracker's POST can land AFTER this one. A terminal record on the
/// `tool_use_id` is what [`on_dispatch_locked`]'s existing idempotence check
/// then finds, so the late observation records nothing rather than resurrecting
/// the claim this call just released. The lock is what makes those two orders
/// converge instead of one of them deciding.
///
/// Callers: the deny arm of both shared-tree routes. The granted-worktree route
/// was added after the 2026-09-13 recurrence, where its denies leaked (#7487).
/// The guard's rewrite there changes `input.isolation` and never `tool_use_id`,
/// so this finds the record the tracker wrote from the original payload.
/// Test: `a_late_observation_of_a_denied_dispatch_records_nothing`,
/// `a_denied_dispatch_cancels_a_record_the_tracker_already_wrote`,
/// `a_dispatch_the_grant_path_denies_records_no_claim_7487`.
pub fn release_denied_dispatch(state: &DaemonState, session: SessionId, payload: &Value) -> bool {
    let Some(tool_use_id) = field(payload, "tool_use_id") else {
        return false;
    };
    let _guard = state.dispatch_record_guard();
    if let Some(id) =
        state.find_delegation(session, |d| d.tool_use_id.as_deref() == Some(tool_use_id))
    {
        return state.mutate_delegation(id, |d| {
            if !d.status.is_terminal() {
                d.status = DelegationStatus::Cancelled;
                d.ended_at = Some(Utc::now());
            }
        });
    }
    let input = payload.get("input");
    let agent = input
        .and_then(|i| i.get("subagent_type"))
        .and_then(Value::as_str)
        .unwrap_or("unknown");
    let task = input
        .and_then(|i| i.get("description"))
        .and_then(Value::as_str)
        .unwrap_or_default();
    let mut delegation = Delegation::observed(session, agent, task, Some(tool_use_id.to_string()));
    delegation.cwd = field(payload, "cwd").map(std::path::PathBuf::from);
    delegation.isolation = dispatch_isolation(input).map(str::to_string);
    delegation.status = DelegationStatus::Cancelled;
    delegation.ended_at = Some(Utc::now());
    state.upsert_delegation(delegation);
    true
}

/// The tool name, when this payload names a subagent-dispatch tool.
///
/// Why: the hot-path early-out — one map lookup and one exact string compare.
/// Test: `ignores_unrelated_tools`.
fn dispatch_tool(payload: &Value) -> Option<&str> {
    payload
        .get("tool")
        .and_then(Value::as_str)
        .filter(|t| is_subagent_dispatch_tool(t))
}

/// Read a `&str` field from a payload, treating empty as absent.
fn field<'a>(payload: &'a Value, key: &str) -> Option<&'a str> {
    payload
        .get(key)
        .and_then(Value::as_str)
        .filter(|s| !s.is_empty())
}

/// The `agentId` a `tool_response` actually offers as a usable correlation
/// handle: present, a string, and non-empty.
///
/// Why (#4163): [`classify_dispatch`]'s presence-check and [`on_launched`]'s
/// consumption used to ask different questions —
/// `r.get("agentId").is_some()` treated `null` and `""` as "we have an id",
/// while the consumer's `.and_then(Value::as_str)` (for `null`) or
/// [`on_subagent_stop`]'s `field` (for `""`) both reject them. A response
/// carrying either shape took the `Launched` branch, pinning the delegation
/// `Running` with an `agent_id` no `SubagentStop` can ever quote back —
/// burning the full 6 h staleness window as a phantom in-flight entry. This
/// is a thin wrapper around [`field`] — the exact same extraction
/// [`on_subagent_stop`] already uses for `agent_id` — rather than an
/// independent body, so all three sites route through one implementation and
/// are structurally incapable of disagreeing again.
/// Test: `null_agent_id_does_not_launch_a_phantom_delegation`,
/// `empty_string_agent_id_does_not_launch_a_phantom_delegation`.
fn usable_agent_id(response: &Value) -> Option<&str> {
    field(response, "agentId")
}

/// Record the working tree a subagent is actually running in (#4311).
///
/// Why: trusty-mpm creates no agent worktrees (ADR-0044 decision 4), so it
/// cannot know the path at dispatch time — the harness chooses it. Without the
/// path the tree has no owner, and an owner-unknown worktree is one ADR-0020
/// correctly forbids anything from removing. That is the whole reason 56
/// unowned trees accumulated under `.claude/worktrees` by the 2026-07-29 count.
///
/// The agent reports its own path, and it is the only party that can. Claude
/// Code stamps `agent_id` into every hook payload a SUBAGENT emits (the same
/// marker `pm_guard_fanout` reads to tell a subagent from the PM) and `tm hook`
/// already forwards it, alongside the `cwd` that hook process is standing in.
/// For an isolated agent that cwd IS the harness worktree. Nothing is created,
/// nothing is relocated, and no new hook is installed — this reads an event the
/// daemon already receives and discards.
/// What: matches `payload.agent_id` against the `agent_id` `on_launched`
/// learned, and writes `payload.cwd` to that delegation's `worktree_path`.
/// Writes nothing when the cwd equals the DISPATCHER's `cwd` — that subagent
/// inherited the caller's tree, so there is no child tree to own — and nothing
/// when the value is already recorded, which is what keeps this off the hot
/// path after the first tool call.
///
/// It is retried on every subagent tool call rather than run once, because the
/// first one can lose a race: `agent_id` is taught only by the dispatch's
/// `PostToolUse`, that hook is installed `async: true`, and until it lands there
/// is no record to match against. A registration that misses simply happens on
/// the next call.
///
/// # The sentinel is written FIRST, and a failed write registers nothing
///
/// The in-memory record is what grants
/// [`super::agent_worktree_reap`] authority to run `git worktree remove
/// --force` on that path. The sentinel is the only evidence of that authority
/// that survives a daemon restart (`daemon::state::core` initialises the
/// delegation map empty and never loads one), so registering after a failed
/// write would grant deletion authority whose durable record does not exist —
/// the fail-OPEN branch, and the exact shape of the bug this whole change
/// closes: a worktree that silently becomes unattributable.
///
/// So the order is write-then-register, and a write failure declines the
/// registration. That is affordable precisely because this function is
/// retried on every subagent tool call: a transient failure self-heals on the
/// next one at no extra cost. Declining leaves the tree owner-UNKNOWN, which
/// is the pre-#4311 state and is never auto-deleted by anything — fail-closed
/// in both directions. It is never fatal to the hook: [`observe`] is
/// observational and must never affect the tool verdict, so the failure is
/// logged at `error!` and the hook returns normally.
///
/// # `cwd` is agent-controlled, so two gates stand in front of the write
///
/// The `cwd` here is `std::env::current_dir()` of the `tm hook` process
/// (`core::standalone::misc`), which is wherever the agent last `cd`-ed. It is
/// not a path trusty-mpm chose, and an agent that visits its own main checkout,
/// `/private/tmp`, or a peer's worktree reports that instead. Writing there
/// would truncate whatever sentinel already sat at that path and retarget the
/// reap at a directory the agent merely walked through.
///
/// So [`is_harness_agent_worktree`] must accept the path — the identical
/// predicate [`super::agent_worktree_reap::reap_worktree`] uses as its own gate,
/// reused rather than restated so the write and the delete cannot disagree
/// about what a worktree is — AND the path's existing sentinel must be either
/// absent/unparsable ([`SentinelOwner::Unknown`], nothing to overwrite) or
/// already this same agent's. Anything else refuses and logs, exactly like the
/// write-failure arm below.
/// # It also tracks where the agent IS, not only what it owns (#6556)
///
/// `worktree_path` is a latch by design — the tree stays this delegation's to
/// own after the agent walks out of it, which is what
/// [`super::agent_worktree_reap`] needs. The shared-tree guard asks a different
/// question ("where is this agent writing now") and a latch cannot answer it, so
/// every subagent hook also refreshes
/// [`Delegation::last_agent_cwd`](crate::core::agent::Delegation::last_agent_cwd).
/// The write happens only when the value changed, so the steady state costs one
/// scan and no allocation.
/// Test: `subagent_tool_call_registers_its_worktree`,
/// `subagent_sharing_the_dispatchers_tree_registers_nothing`,
/// `subagent_tool_calls_track_the_current_working_directory`,
/// `an_unknown_agent_id_registers_nothing`,
/// `a_failed_sentinel_write_registers_no_worktree`,
/// `a_cwd_outside_the_harness_store_gets_no_sentinel`,
/// `a_cwd_owned_by_another_agent_is_never_overwritten`,
/// `a_cwd_owned_by_a_managed_session_is_never_overwritten`.
/// Why this agent may NOT claim `cwd` as its worktree, or `None` if it may.
///
/// Why: the `cwd` a subagent reports is `std::env::current_dir()` of its own
/// `tm hook` process, so it follows wherever the agent `cd`-ed. It is a value
/// the agent controls, and a claim on it grants the reaper authority to delete
/// that directory. Four independent claims must all hold, and each answers a
/// question the others cannot.
/// What: `Some(reason)` refuses and is logged; `None` proceeds to the write.
///
/// 1. **Shape** — [`is_harness_agent_worktree`], the identical predicate
///    [`super::agent_worktree_reap::reap_worktree`] gates on, imported rather
///    than restated so the write and the delete cannot disagree about what a
///    worktree is. Rejects a main checkout, a scratchpad, a nested path.
/// 2. **Project** — the store this path sits in must be the store of the
///    DISPATCHING session's own checkout. Claim 1 has already established the
///    path is `<X>/.claude/worktrees/<name>`, so `<X>` is the checkout owning
///    that store; the dispatcher's `cwd` must be `<X>` or sit under it. Under
///    [ADR-0037](../../../../../docs/adr/0037-pm-placement-precedence-main-checkout-by-default.md)
///    a PM is normally in the main checkout, which IS `<X>`, and under
///    [ADR-0036](../../../../../docs/adr/0036-all-worktrees-are-siblings-under-claude-worktrees.md)
///    a PM working from a worktree is a flat sibling beneath the same `<X>` —
///    both normal placements pass. This is deliberately LEXICAL rather than a
///    `harness_root_for` call: that resolves through `git rev-parse`, and this
///    runs inside the synchronous `PreToolUse` budget. The lexical form refuses
///    a superset (a PM in a different checkout of the same project), which is
///    the fail-closed direction.
/// 3. **Sentinel** — the path's existing sentinel must be absent/unparsable or
///    already this same agent's. `fs::write` truncates, so without this an
///    agent reporting a peer's tree erases that peer's ownership record, after
///    which the peer's tree reaps on the wrong agent's exit and the peer's own
///    exit reaps nothing.
/// 4. **Registration** — no OTHER non-terminal delegation may already record
///    this path. Claim 3 cannot see a peer worktree that carries no sentinel,
///    which is every directory in the store until they acquire one; this claim
///    covers exactly that case from the daemon's own live state, and is the
///    write-side mirror of the reap's in-use gate.
///
/// Residual, stated: a sentinel-LESS peer whose delegation was also lost to a
/// daemon restart is invisible to claims 3 and 4 both. Such a directory is
/// unattributed and unregistered, so nothing else can reclaim it either; the
/// exposure closes as worktrees acquire sentinels.
///
/// A convention that WOULD close it — the harness names each tree
/// `agent-<agent_id>`, which holds for all 16 such directories on this machine
/// — is deliberately not used. Nothing in this codebase or in any contract with
/// Claude Code establishes that name, so a harness rename would silently stop
/// every registration rather than failing visibly.
/// Test: `a_cwd_outside_the_harness_store_gets_no_sentinel`,
/// `a_cwd_in_another_projects_store_is_refused`,
/// `a_cwd_owned_by_another_agent_is_never_overwritten`,
/// `a_cwd_owned_by_a_managed_session_is_never_overwritten`,
/// `a_cwd_another_live_delegation_holds_is_refused`,
/// `an_agent_may_rewrite_its_own_sentinel`.
fn claim_refused(
    state: &DaemonState,
    session: SessionId,
    agent_id: &str,
    cwd: &std::path::Path,
) -> Option<String> {
    if !is_harness_agent_worktree(cwd) {
        return Some("it is not a `.claude/worktrees/<name>` leaf".to_string());
    }
    // `<X>/.claude/worktrees/<name>` — three parents up is `<X>`, the checkout
    // that owns the store. Claim 1 guarantees the first two exist.
    let store_owner = cwd
        .parent()
        .and_then(std::path::Path::parent)
        .and_then(std::path::Path::parent)?;
    let dispatcher_cwd = state
        .delegations_for(session)
        .into_iter()
        .find(|d| d.agent_id.as_deref() == Some(agent_id))
        .and_then(|d| d.cwd);
    match dispatcher_cwd {
        Some(dispatch_from) if dispatch_from.starts_with(store_owner) => {}
        Some(dispatch_from) => {
            return Some(format!(
                "its store belongs to {}, but the dispatching session works from {} — an \
                 agent may not claim a worktree in another project's store",
                store_owner.display(),
                dispatch_from.display()
            ));
        }
        // No recorded dispatcher cwd is no proof of a shared project.
        None => {
            return Some(
                "the dispatching delegation records no working directory, so nothing \
                 establishes this store belongs to its project"
                    .to_string(),
            );
        }
    }
    match read_sentinel_owner(cwd) {
        SentinelOwner::Unknown => {}
        SentinelOwner::Agent(existing, _) if existing.agent_id == agent_id => {}
        occupied => return Some(format!("{occupied:?} already owns it")),
    }
    if let Some(holder) = state.all_delegations().into_iter().find(|d| {
        !d.status.is_terminal()
            && d.agent_id.as_deref() != Some(agent_id)
            && d.worktree_path.as_deref() == Some(cwd)
    }) {
        return Some(format!(
            "agent {} is still running and already registers it",
            holder.agent_id.as_deref().unwrap_or("<unknown>")
        ));
    }
    None
}

fn register_agent_worktree(state: &DaemonState, session: SessionId, payload: &Value) {
    let Some(agent_id) = field(payload, "agent_id") else {
        return;
    };
    let Some(cwd) = field(payload, "cwd").map(std::path::PathBuf::from) else {
        return;
    };
    let Some(id) = state.find_delegation(session, |d| d.agent_id.as_deref() == Some(agent_id))
    else {
        return;
    };
    // #6556 critic round: track where this agent is NOW, in the same pass that
    // decides whether to claim. `worktree_path` is a latch and must stay one —
    // the reap needs it — so the current directory is its own field. Written
    // only when it changed, which keeps the steady state off the hot path.
    let mut claim_needed = false;
    state.mutate_delegation(id, |d| {
        if d.last_agent_cwd.as_deref() != Some(cwd.as_path()) {
            d.last_agent_cwd = Some(cwd.clone());
        }
        claim_needed = d.worktree_path.as_deref() != Some(cwd.as_path())
            && d.cwd.as_deref() != Some(cwd.as_path());
    });
    if !claim_needed {
        return;
    }
    if let Some(refusal) = claim_refused(state, session, agent_id, &cwd) {
        tracing::warn!(
            agent_id,
            cwd = %cwd.display(),
            "delegation: not claiming this worktree — {refusal} (#4311)"
        );
        return;
    }
    let owner = AgentWorktreeOwner {
        agent_id: agent_id.to_string(),
        delegation_id: id,
        parent_session_id: session,
    };
    if let Err(e) = write_agent_sentinel(&cwd, owner) {
        tracing::error!(
            agent_id,
            worktree = %cwd.display(),
            "delegation: could not write the agent worktree's ownership sentinel ({e}); \
             leaving the tree unregistered and owner-unknown rather than granting a reap \
             authority nothing on disk records — retried on this agent's next tool call (#4311)"
        );
        return;
    }
    tracing::info!(
        agent_id,
        worktree = %cwd.display(),
        "delegation: registered the agent worktree its dispatch was granted (#4311)"
    );
    state.mutate_delegation(id, |d| d.worktree_path = Some(cwd.clone()));
}

/// `PreToolUse` on a dispatch tool: a subagent is starting now.
///
/// Why: this is the moment tracking must record a *live* child, so a PM with
/// work in flight is never mistaken for idle.
/// What: reads `input.subagent_type` (agent) and `input.description` (task).
/// Idempotent on `tool_use_id` — a redelivered hook updates nothing twice. Tries
/// [`dedup`] first so a matching `agent_delegate` record is promoted in place to
/// `Running`; otherwise inserts a fresh
/// [`Delegation::observed`](crate::core::agent::Delegation::observed). Records
/// `cwd`, `isolation` (#4480) and `transcript_path` when present.
/// Test: `pre_tool_use_creates_running_delegation`,
/// `dedups_declaration_and_observation`, `duplicate_pre_tool_use_is_idempotent`,
/// `pre_tool_use_records_declared_isolation`.
fn on_dispatch(state: &DaemonState, session: SessionId, payload: &Value) {
    let _guard = state.dispatch_record_guard();
    on_dispatch_locked(state, session, payload);
}

/// Correct an existing delegation's `isolation`, or create the record (#5769).
///
/// Why: `tm hook --pm-guard` rewrites an unisolated dispatch made from a main
/// checkout into an isolated one (ADR-0048 decision 1), but [`on_dispatch`]
/// observes the ORIGINAL payload, so the record read `isolation: None` — naming
/// an agent that had just been moved into its own worktree as a live writer in
/// the shared checkout. Decision 10 then denies a `git merge` or `git rebase`
/// there on that record,
/// for the six hours of `RUNNING_STALE_AFTER_SECS`, which blocks the release
/// flow's fast-forward of the main checkout.
///
/// It is an UPSERT and not a second call to [`observe`] for two reasons read
/// from this file. [`on_dispatch`] returns early when a delegation with the same
/// `tool_use_id` already exists, so an `observe`-based grant would correct
/// nothing whenever the tracker's hook won the race — and the two hooks fire on
/// the same event, so which one wins is not decidable here. And the route the
/// guard would otherwise post to re-derives eligibility and rejects an isolated
/// dispatch by design. So isolation must be written over whatever landed first,
/// which makes the two arrival orders converge instead of one of them deciding.
/// What: with the dispatch-record lock held, sets `isolation` on the delegation
/// carrying `payload.tool_use_id`, or creates the record from this payload when
/// none exists — in which case the tracker's own later hook is a no-op on
/// `tool_use_id` and the granted isolation survives.
///
/// Two payloads record NOTHING and return `false`, and both rules live HERE
/// rather than in the caller. One with no `tool_use_id`: without that key the
/// record could not be found again, so creating one would leave a second,
/// unisolated record beside the tracker's — the phantom duplicated rather than
/// removed. And one whose `isolation` does not
/// [`isolation_separates_working_tree`]: this function's whole purpose is to
/// erase a record that names a writer as sharing the checkout, so writing a
/// non-separating mode over a correct record would write the very phantom it
/// exists to remove. The route that calls this re-derives the same test to
/// decide eligibility; keeping it in the writer as well means a second caller,
/// or a refactor of that route, cannot reach the write without it.
///
/// A record found in a NON-LIVE status is revived to `Running` with `ended_at`
/// cleared (#7487 critic round). The only caller reaches this from inside
/// [`crate::daemon::state::DaemonState::claim_shared_tree_dispatch`]'s admission
/// arm, so the dispatch is running; a deny of the same `tool_use_id` moments
/// earlier left a `Cancelled` tombstone, and writing `isolation` onto it without
/// the revival described a live agent as Ended — which the reclaim reads as
/// permission to delete its worktree (#5661).
/// Test: `a_grant_and_the_tracker_converge_in_either_order`,
/// `record_granted_isolation_refuses_a_non_separating_mode`,
/// `an_admitted_grant_revives_the_record_its_own_deny_tombstoned_7487`.
pub fn record_granted_isolation(state: &DaemonState, session: SessionId, payload: &Value) -> bool {
    let Some(tool_use_id) = field(payload, "tool_use_id") else {
        return false;
    };
    let declared = dispatch_isolation(payload.get("input"));
    if !isolation_separates_working_tree(declared) {
        return false;
    }
    let Some(isolation) = declared.map(str::to_string) else {
        return false;
    };
    let _guard = state.dispatch_record_guard();
    if let Some(id) =
        state.find_delegation(session, |d| d.tool_use_id.as_deref() == Some(tool_use_id))
    {
        state.mutate_delegation(id, |d| {
            // #7487: this call runs only from inside
            // `claim_shared_tree_dispatch`'s ADMISSION arm, so by the time it is
            // reached the daemon has decided this dispatch RUNS. A record found
            // in a non-live status is therefore a tombstone the SAME id already
            // earned — a deny of the unisolated form that the guard then re-made
            // as a grant — and setting only `isolation` over it left a live
            // agent described as Ended. The reclaim reads that and can delete
            // the worktree out from under it, which is #5661's shape.
            if !d.status.is_live() {
                d.status = DelegationStatus::Running;
                d.ended_at = None;
                d.started_at = Some(Utc::now());
            }
            d.isolation = Some(isolation.clone());
        });
        return true;
    }
    on_dispatch_locked(state, session, payload);
    true
}

/// [`on_dispatch`] with the dispatch-record lock already held.
///
/// Why: [`record_granted_isolation`] holds that lock across its own
/// find-then-insert and then falls through to this body, so the locking cannot
/// live here — it would be a second, non-reentrant acquisition of a
/// `parking_lot::Mutex`, which deadlocks rather than failing.
fn on_dispatch_locked(state: &DaemonState, session: SessionId, payload: &Value) {
    let input = payload.get("input");
    let agent = input
        .and_then(|i| i.get("subagent_type"))
        .and_then(Value::as_str)
        .unwrap_or("unknown");
    let task = input
        .and_then(|i| i.get("description"))
        .and_then(Value::as_str)
        .unwrap_or_default();
    let tool_use_id = field(payload, "tool_use_id").map(str::to_string);

    // Idempotence: an already-tracked tool_use_id means this hook was redelivered.
    if let Some(id) = tool_use_id.as_deref()
        && state
            .find_delegation(session, |d| d.tool_use_id.as_deref() == Some(id))
            .is_some()
    {
        return;
    }

    let cwd = field(payload, "cwd").map(std::path::PathBuf::from);
    let transcript = field(payload, "transcript_path").map(std::path::PathBuf::from);
    // #4480: what the dispatch declared, recorded verbatim. Absent is the
    // default and the hazardous case — the subagent inherits `cwd`.
    let isolation = dispatch_isolation(input).map(str::to_string);

    // Merge with a matching `agent_delegate` declaration when one is pending.
    if let Some(existing) = dedup(state, session, agent, task) {
        state.mutate_delegation(existing, |d| {
            d.status = DelegationStatus::Running;
            d.started_at = Some(Utc::now());
            d.tool_use_id = tool_use_id.clone();
            d.cwd = cwd.clone();
            d.isolation = isolation.clone();
            d.transcript_path = transcript.clone();
            if d.task.is_empty() && !task.is_empty() {
                d.task = task.to_string();
            }
        });
        return;
    }

    let mut delegation = Delegation::observed(session, agent, task, tool_use_id);
    delegation.cwd = cwd;
    delegation.isolation = isolation;
    delegation.transcript_path = transcript;
    state.upsert_delegation(delegation);
}

/// Find a pending `agent_delegate` declaration this dispatch should merge into.
///
/// Why: a PM instructed to call `agent_delegate` *and* actually dispatch would
/// otherwise produce two records for one subagent, inflating the live-children
/// count and leaving an immortal `Queued` orphan (nothing would ever terminalize
/// the declaration, since only the observed record carries the correlation
/// keys).
/// What: the most recent same-session, same-agent delegation that is still
/// `Queued`, was `McpDeclared`, has not already been bound to a `tool_use_id`,
/// was created within [`DEDUP_WINDOW_SECS`], and whose task [`tasks_match`]es
/// this dispatch's description.
///
/// The agent name alone is NOT a sufficient key (#2864 review, MEDIUM 2): this
/// repo dispatches several `rust-engineer`s concurrently, so "same agent,
/// recently" false-merges two distinct dispatches — halving the live-child count
/// and leaving the merged record showing the *declaration's* task text, which is
/// precisely the string `/tm-session-pause` puts in front of a human. When in
/// doubt this now declines to merge: two records for one subagent overcounts and
/// is corrected by the staleness sweep, whereas a false merge undercounts work
/// in flight and mislabels it, with no recovery.
/// Test: `dedups_declaration_and_observation`, `dedup_window_expires`,
/// `dedup_ignores_different_agent`, `dedup_does_not_merge_a_different_task`.
fn dedup(state: &DaemonState, session: SessionId, agent: &str, task: &str) -> Option<DelegationId> {
    let cutoff = Utc::now() - chrono::Duration::seconds(DEDUP_WINDOW_SECS);
    state.latest_delegation_matching(session, |d| {
        d.source == DelegationSource::McpDeclared
            && d.status == DelegationStatus::Queued
            && d.tool_use_id.is_none()
            && d.agent == agent
            && d.created_at >= cutoff
            && tasks_match(&d.task, task)
    })
}

/// Do a declared task and a dispatched description describe the same work?
///
/// Why: the task text is the only discriminator available between two dispatches
/// to the same agent inside the dedup window, and it must tolerate the PM
/// wording the `agent_delegate` task and the dispatch `description` slightly
/// differently while still rejecting two plainly different work items.
/// What: case- and whitespace-insensitive prefix match in either direction
/// (which subsumes equality), so a short dispatch description matches the longer
/// declaration it summarises.
///
/// The two empty cases are NOT symmetric (#2864 re-review, LOW 1). A dispatch
/// with no description offers nothing to discriminate on, and merging it would
/// keep the *declaration's* text as the label for a dispatch we cannot identify
/// — precisely the mislabel this discriminator exists to prevent — so it
/// declines. An unlabelled *declaration* is different: merging adopts the
/// dispatch's own text (`on_dispatch` fills an empty task), so no wrong label
/// can survive, and it merges.
/// Test: `dedups_declaration_and_observation`,
/// `dedup_does_not_merge_a_different_task`,
/// `dedup_declines_a_description_less_dispatch`.
fn tasks_match(declared: &str, dispatched: &str) -> bool {
    if dispatched.is_empty() {
        return false;
    }
    if declared.is_empty() {
        return true;
    }
    let normalize = |s: &str| {
        s.split_whitespace()
            .collect::<Vec<_>>()
            .join(" ")
            .to_lowercase()
    };
    let (a, b) = (normalize(declared), normalize(dispatched));
    a.starts_with(&b) || b.starts_with(&a)
}

/// `PostToolUse` on a dispatch tool: learn `agentId`, terminalize only if the
/// dispatch really returned.
///
/// Why: see the module note — an async dispatch returns in ~1 ms while the
/// subagent runs on, so this is primarily the *join* that binds `tool_use_id` to
/// the `agent_id` that `SubagentStop` will quote.
/// What: locates the delegation by `tool_use_id`, stores
/// `tool_response.agentId`, and refines the tier from `resolvedModel`. Whether
/// to terminalize is decided entirely by [`classify_dispatch`], which is a
/// three-way answer — `Launched` / `Returned` / `Unknown` — not a boolean.
/// Test: `async_launch_keeps_delegation_running`,
/// `synchronous_post_tool_use_completes_delegation`,
/// `post_tool_use_failure_marks_failed`,
/// `post_tool_use_without_tool_response_stays_running`,
/// `post_tool_use_with_unrecognized_response_stays_running`,
/// `post_tool_use_with_only_an_agent_id_stays_running`,
/// `changed_async_status_value_with_an_agent_id_stays_running`,
/// `null_agent_id_does_not_launch_a_phantom_delegation`,
/// `empty_string_agent_id_does_not_launch_a_phantom_delegation`.
fn on_launched(state: &DaemonState, session: SessionId, payload: &Value, event: HookEvent) {
    let Some(tool_use_id) = field(payload, "tool_use_id") else {
        return;
    };
    let Some(id) =
        state.find_delegation(session, |d| d.tool_use_id.as_deref() == Some(tool_use_id))
    else {
        return;
    };

    // Only a response we actually recognized carries evidence either way.
    let response = payload
        .get("tool_response")
        .filter(|r| recognized_response(r));
    // #4163: use the same extraction `classify_dispatch` uses for its
    // presence-check, so a null/empty `agentId` is never stored as a handle
    // nothing can ever resolve.
    let agent_id = response.and_then(usable_agent_id).map(str::to_string);
    let tier = response
        .and_then(|r| r.get("resolvedModel"))
        .and_then(Value::as_str)
        .map(ModelTier::from_model_id);

    // The dispatch call itself failed: no subagent was left running.
    let errored = event == HookEvent::PostToolUseFailure
        || response
            .and_then(|r| r.get("is_error"))
            .and_then(Value::as_bool)
            .unwrap_or(false);
    let outcome = classify_dispatch(response);

    // #4142: the stop for this agent may already have arrived and found nothing
    // to resolve. `peek` rather than take — the entry is cleared below only once
    // the terminalization it authorises has actually landed, so a mutate that
    // finds no record leaves the stop recoverable instead of discarding it.
    let pending = agent_id
        .as_deref()
        .and_then(|a| state.pending_stops().peek(session, a));

    let mutated = state.mutate_delegation(id, |d| {
        if agent_id.is_some() {
            d.agent_id = agent_id.clone();
        }
        if let Some(t) = tier {
            d.tier = t;
        }
        // Fail CLOSED. `Launched` never terminalizes, whatever else the event
        // says — a running subagent outranks an errored dispatch call, and
        // `SubagentStop` will close it. `Unknown` terminalizes only on an
        // explicit dispatch failure, which is itself positive evidence that
        // nothing was left running.
        //
        // #4142: a stop already observed for this exact `agent_id` outranks all
        // three. It is the authoritative terminal signal, and applying the
        // status it determined is what makes the two arrival orders converge on
        // one record rather than one of them deciding.
        let terminal = pending.or(match outcome {
            DispatchOutcome::Launched => None,
            DispatchOutcome::Returned => Some(if errored {
                DelegationStatus::Failed
            } else {
                DelegationStatus::Completed
            }),
            DispatchOutcome::Unknown if errored => Some(DelegationStatus::Failed),
            DispatchOutcome::Unknown => None,
        });
        if let Some(status) = terminal {
            d.status = status;
            d.ended_at = Some(Utc::now());
        }
    });

    // Only now is the deferred stop spent. A `false` here means the record
    // vanished between the lookup and the write, so the stop stays pending for
    // whatever record carries this id next.
    if mutated
        && pending.is_some()
        && let Some(a) = agent_id.as_deref()
    {
        state.pending_stops().clear(session, a);
    }
}

/// What a dispatch `PostToolUse` response says about whether a subagent is
/// still running.
///
/// Why this is three-valued and not a boolean: "I could parse this response"
/// and "this response says nothing was left running" are different
/// propositions, and the first round of #2864 review was about exactly that
/// conflation one layer up. `Unknown` is a first-class answer here for the same
/// reason [`DelegationStatus::Stale`] is a first-class status: not knowing is
/// not the same as knowing it finished.
/// Test: the `post_tool_use_*` suite.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DispatchOutcome {
    /// A subagent is running now — an explicit async marker, or an `agentId`
    /// handle for a future `SubagentStop` to quote.
    Launched,
    /// The call returned rather than launching; nothing is left running.
    Returned,
    /// The response is absent or unrecognized — no conclusion is available.
    Unknown,
}

/// Classify a dispatch response (#2864 review HIGH 1, re-review MEDIUM 1).
///
/// Why: an absent response used to read as "not async, therefore finished",
/// terminalizing a still-running subagent ~1 ms after launch — and absence is
/// not hypothetical, since `tm hook`'s S1 projection ([`TOOL_RESPONSE_KEYS`])
/// omits `tool_response` entirely when the payload is not an object or carries
/// none of those keys. That is [`DispatchOutcome::Unknown`].
///
/// The re-review then found a narrower conflation: a response holding only
/// `agentId` produced `Completed` *and* stored the `agent_id` in the same
/// breath — recording "expect a `SubagentStop` for this agent" and "this
/// already finished" simultaneously. Hence the launch test comes first and
/// includes `agentId`.
///
/// What, in order:
///
/// - `Launched` — `isAsync == true`, `status == "async_launched"`, or a
///   *usable* `agentId` ([`usable_agent_id`]: non-null, non-empty) is present.
///   An `agentId` is by design the *async* correlation key; handing one back
///   is evidence of a launch, never of a return. Testing it first is also
///   what makes a `status` whose **value** drifted (rather than its key)
///   safe, since such a response still carries its `agentId`. A `null` or
///   `""` `agentId` is not a usable handle (#4163) — nothing downstream could
///   ever resolve it — so it does not take this branch.
/// - `Returned` — any other recognized response.
/// - `Unknown` — absent, not an object, or carrying no [`TOOL_RESPONSE_KEYS`]
///   member.
///
/// # KNOWN LIMITATION — a response with no usable `agentId` is read as a return
///
/// `Returned` is deliberately still *residual* rather than affirmative
/// (`isAsync == false` or a non-launch `status`), so a recognized response that
/// is silent about liveness — `resolvedModel` alone, `is_error` alone — is
/// treated as a synchronous return. That is a real fail-open band and it is
/// **knowingly left open** here, because closing it would make things worse
/// today, not better:
///
/// [`on_subagent_stop`] resolves a delegation *only* by `agent_id`, and
/// `agent_id` is taught *only* by this function. A response with no *usable*
/// `agentId` — absent, `null`, `""`, or any non-string ([`usable_agent_id`]) —
/// therefore has **no** route to termination other than this branch, because
/// nothing can ever quote it back to us. Making this branch affirmative would
/// convert a rare fail-open into a guaranteed six-hour phantom "agent in flight"
/// for every such dispatch.
///
/// The #4142 deferred-stop ledger does NOT close this. That ledger recovers a
/// stop that merely arrived *early*, by holding it until the `agent_id` is
/// taught. When no usable `agentId` is ever taught, there is no key for it to be
/// held against and the delegation is unreachable by any stop, in any order.
///
/// The band is unobserved: Claude Code 2.1.220 returns
/// `{isAsync, status, agentId, resolvedModel}` for a dispatch, which takes the
/// `Launched` branch. Tightening it needs a correlation key this response does
/// not carry, so it is not attempted here.
/// Test: `post_tool_use_without_tool_response_stays_running`,
/// `post_tool_use_with_unrecognized_response_stays_running`,
/// `post_tool_use_with_only_an_agent_id_stays_running`,
/// `changed_async_status_value_with_an_agent_id_stays_running`,
/// `synchronous_post_tool_use_completes_delegation`,
/// `liveness_silent_response_is_read_as_a_return_known_gap`,
/// `null_agent_id_does_not_launch_a_phantom_delegation`,
/// `empty_string_agent_id_does_not_launch_a_phantom_delegation`.
fn classify_dispatch(response: Option<&Value>) -> DispatchOutcome {
    let Some(r) = response else {
        return DispatchOutcome::Unknown;
    };
    if r.get("isAsync").and_then(Value::as_bool) == Some(true)
        || r.get("status").and_then(Value::as_str) == Some("async_launched")
        // #4163: was `r.get("agentId").is_some()` — key presence only, so a
        // `null` or `""` agentId took this branch even though nothing
        // downstream could ever resolve it. Use the same extraction the
        // consumer in `on_launched` uses.
        || usable_agent_id(r).is_some()
    {
        return DispatchOutcome::Launched;
    }
    DispatchOutcome::Returned
}

/// Is this `tool_response` one we can draw a conclusion from?
///
/// Why: `tm hook`'s S1 projection already restricts the forwarded object to
/// [`TOOL_RESPONSE_KEYS`], but the daemon must not depend on a filter living in
/// a different crate target for its fail-direction. Re-asserting it here means a
/// more permissive S1 (a raw string, a content-block array) cannot silently turn
/// "unrecognized" back into "understood".
/// What: `true` iff `response` is an object carrying at least one
/// [`TOOL_RESPONSE_KEYS`] member.
/// Test: `post_tool_use_with_unrecognized_response_stays_running`.
fn recognized_response(response: &Value) -> bool {
    response
        .as_object()
        .is_some_and(|o| TOOL_RESPONSE_KEYS.iter().any(|k| o.contains_key(*k)))
}

/// `SubagentStop`: terminalize exactly the delegation that ended, or remember
/// the stop until the delegation can be found (#4142).
///
/// Why: this is the authoritative completion signal, and the only one that
/// arrives when the subagent actually finishes. It resolves only by `agent_id`,
/// which only [`on_launched`] teaches — and `PostToolUse` is installed
/// `async: true` while this event is synchronous, so a stop can reach the daemon
/// before the id it quotes has been recorded. Discarding it there left the
/// delegation `Running` for the full six-hour staleness window as a phantom
/// in-flight agent.
/// What: `Failed` for `SubagentStopFailure`, else `Completed`. Three
/// dispositions, and the third is the #4142 fix:
///
/// 1. A **live** delegation carries this `agent_id` — terminalize it. `Stale` is
///    deliberately not terminal, so a stop arriving after the staleness sweep
///    gave up still replaces "we lost track" with the truth.
/// 2. A **terminal** delegation carries it — this stop was redelivered and its
///    fact is already recorded. Nothing to do, and nothing to remember.
/// 3. **No delegation carries it at all** — either the id is not ours, or its
///    `PostToolUse` has not landed yet, and the two are indistinguishable here.
///    So the stop goes to [`crate::daemon::state::pending_stops`] and
///    [`on_launched`] applies it the moment the id becomes known. A stop that
///    was never ours simply expires there.
///
/// A missing `agent_id` still terminalizes nothing and records nothing — with no
/// correlation key there is nothing to defer, and the module note explains why a
/// "most recent" guess is forbidden.
///
/// # Disposition 3 also reconciles by `agent_type` (#6556)
///
/// Deferring is the whole answer only when the `agent_id` will eventually be
/// taught. When the dispatch's `PostToolUse` is lost outright — it is installed
/// `async: true`, and its POST fails open on a 2 s budget — no id is ever
/// learned, nothing can consume the deferred stop, and the record reads live for
/// the six hours of `RUNNING_STALE_AFTER_SECS`. Observed 2026-09-01: records for
/// a `version-control` agent and a `rust-engineer` outlived both agents by ~80
/// minutes and four ADR-0049 documents-only commits were denied naming them.
/// #6497 fixed the sibling case where the owning SESSION died; this is the case
/// where the agent finished normally.
///
/// So before deferring, [`reconcile_by_agent_type`] looks for the one live
/// record this stop can only be about. It is driven by a real stop signal —
/// positive evidence that a subagent of this session, of this type, has ended
/// (ADR-0045) — declines whenever more than one record could match, and writes
/// [`DelegationStatus::Stale`] rather than a terminal status, because the
/// identification is by type rather than by id. The stop is STILL deferred, so a
/// `PostToolUse` that lands late replaces "we lost track" with the truth.
/// Test: `subagent_stop_completes_matching_delegation`,
/// `subagent_stop_without_agent_id_terminalizes_nothing`,
/// `concurrent_delegations_terminalize_independently`,
/// `out_of_order_subagent_stop_resolves_when_its_post_tool_use_lands`,
/// `a_redelivered_stop_for_a_finished_delegation_records_no_pending_stop`,
/// `a_completed_agents_record_stops_blocking_when_its_id_was_never_taught`.
fn on_subagent_stop(state: &DaemonState, session: SessionId, payload: &Value, event: HookEvent) {
    let Some(agent_id) = field(payload, "agent_id") else {
        tracing::trace!(
            "SubagentStop without agent_id — no delegation terminalized (#2864); a \
             'most recent' guess would close the wrong one under concurrency"
        );
        return;
    };
    let status = if event == HookEvent::SubagentStopFailure {
        DelegationStatus::Failed
    } else {
        DelegationStatus::Completed
    };
    if let Some(id) = state.find_delegation(session, |d| {
        d.agent_id.as_deref() == Some(agent_id) && !d.status.is_terminal()
    }) {
        state.terminate_delegation(id, status);
        return;
    }
    if state
        .find_delegation(session, |d| d.agent_id.as_deref() == Some(agent_id))
        .is_some()
    {
        return;
    }
    // #6556: no record carries this id, and one may never learn it. Stale the
    // single record this stop can only be about, before deferring it.
    reconcile_by_agent_type(state, session, payload);
    // #4142: the id is unknown to this session — hold the stop rather than drop it.
    state
        .pending_stops()
        .record(session, agent_id, status, Utc::now());
}

/// Stale the one live record a stop with an unresolvable `agent_id` can only be
/// about (#6556).
///
/// Why: see [`on_subagent_stop`]'s disposition 3. A `SubagentStop` whose id no
/// record carries still proves that a subagent of this session ENDED, and
/// `agent_type` names which kind. When exactly one live record of that kind has
/// never learned an `agent_id`, that record is the only thing the stop can
/// describe, so leaving it live is a claim this daemon has evidence against.
///
/// What: with the dispatch-record lock held — the same lock every other writer
/// of a delegation's identity fields takes, so a `PostToolUse` cannot teach an
/// id between the count and the write — collects the session's live delegations
/// whose `agent` equals `payload.agent_type` and whose `agent_id` is `None`, and
/// writes [`DelegationStatus::Stale`] when there is EXACTLY ONE. Returns whether
/// it wrote.
///
/// Three refusals, each deliberate:
///
/// * **No `agent_type`** — nothing to match on, so nothing is reconciled. A
///   match on liveness alone would be the "most recent guess" the module note
///   forbids.
/// * **More than one candidate** — two unresolved agents of one type are
///   indistinguishable here, and closing the wrong one would stop the ADR-0048
///   guard counting a writer that is still in the tree. Ambiguity declines.
/// * **`Stale`, not `Completed`** — the identification is by TYPE. A stop whose
///   own `PreToolUse` was lost could name a record belonging to a sibling of the
///   same type, and `Stale` is the status that says "tracking gave up" without
///   asserting the agent finished. It is non-terminal, so the deferred stop
///   still resolves it to the truth if the id is taught later, and it is not
///   live, so it stops blocking a dispatch or an ADR-0049 commit immediately —
///   which is the whole closure condition.
///
/// Test: `a_completed_agents_record_stops_blocking_when_its_id_was_never_taught`,
/// `two_unresolved_records_of_one_type_are_left_alone`,
/// `a_stop_without_an_agent_type_reconciles_nothing`,
/// `reconciliation_never_touches_a_record_that_learned_its_id`.
fn reconcile_by_agent_type(state: &DaemonState, session: SessionId, payload: &Value) -> bool {
    let Some(agent_type) = field(payload, "agent_type") else {
        return false;
    };
    let _guard = state.dispatch_record_guard();
    let mut candidates = state
        .delegations_for(session)
        .into_iter()
        .filter(|d| d.status.is_live() && d.agent_id.is_none() && d.agent == agent_type);
    let Some(only) = candidates.next() else {
        return false;
    };
    if candidates.next().is_some() {
        tracing::debug!(
            agent_type,
            "delegation: a stop names an agent type with more than one unresolved record — \
             declining to guess which finished (#6556)"
        );
        return false;
    }
    // #6556 critic round: `stale_by_agent_type` is what keeps this record
    // OCCUPYING the tree for a new file-mutating dispatch while releasing it for
    // an ADR-0049 commit. See `DaemonState::shared_tree_occupants`.
    let wrote = state.mutate_delegation(only.id, |d| {
        d.status = DelegationStatus::Stale;
        d.stale_by_agent_type = true;
    });
    if wrote {
        tracing::info!(
            agent_type,
            delegation = ?only.id,
            "delegation: a subagent of this type stopped and only this record could be it — \
             marking it stale so it stops counting as a live writer (#6556)"
        );
    }
    wrote
}

#[cfg(test)]
#[path = "delegation_tracker_tests.rs"]
mod tests;