trusty-mpm 1.8.2

//! Tests for the shared-tree dispatch route (#4480, #5324).
//!
//! Why: the guard's whole decision rests on this answer, and each filter it
//! applies is a separate way to false-deny the PM. Each gets its own case. Since
//! #5324 the route also CLAIMS the directory, so each way it could claim one it
//! should not have — a read-only agent, an isolated dispatch, a non-dispatch
//! tool, a payload with no directory — is its own case too.
//! Test: this *is* the test module.

use super::*;
use crate::core::agent::{Delegation, DelegationStatus};
use crate::core::paths::FrameworkPaths;

/// Build a hermetic state plus one session id, mirroring `api_tests`.
fn hermetic() -> (Arc<DaemonState>, tempfile::TempDir, SessionId) {
    let dir = tempfile::tempdir().expect("temp dir");
    let paths = FrameworkPaths::under(dir.path());
    let state = Arc::new(DaemonState::with_paths(&paths));
    (state, dir, SessionId(uuid::Uuid::new_v4()))
}

/// Insert one observed delegation with the given shape.
fn insert(
    state: &DaemonState,
    session: SessionId,
    agent: &str,
    cwd: &str,
    isolation: Option<&str>,
    tool_use_id: Option<&str>,
    status: DelegationStatus,
) {
    let mut d = Delegation::observed(session, agent, "task", tool_use_id.map(str::to_string));
    d.cwd = Some(PathBuf::from(cwd));
    d.isolation = isolation.map(str::to_string);
    d.status = status;
    state.upsert_delegation(d);
}

/// One dispatch in the daemon's forwarded hook shape — what the guard POSTs.
fn dispatch(
    cwd: &str,
    agent: &str,
    isolation: Option<&str>,
    tool_use_id: Option<&str>,
) -> SharedTreeDispatchRequest {
    let mut input = serde_json::json!({"subagent_type": agent, "description": "go"});
    if let Some(i) = isolation {
        input["isolation"] = Value::String(i.to_string());
    }
    let mut payload = serde_json::json!({"cwd": cwd, "tool": "Agent", "input": input});
    if let Some(id) = tool_use_id {
        payload["tool_use_id"] = Value::String(id.to_string());
    }
    SharedTreeDispatchRequest { payload }
}

/// Drive the route and unwrap its body.
async fn call(
    state: &Arc<DaemonState>,
    session: SessionId,
    req: SharedTreeDispatchRequest,
) -> SharedTreeWritersResponse {
    let Json(body) =
        shared_tree_dispatch_route(State(state.clone()), Path(session.0.to_string()), Json(req))
            .await
            .expect("route succeeds");
    body
}

/// Drive the granted-worktree route and unwrap its body.
async fn granted_call(
    state: &Arc<DaemonState>,
    session: SessionId,
    req: SharedTreeDispatchRequest,
) -> SharedTreeWritersResponse {
    let Json(body) =
        granted_worktree_route(State(state.clone()), Path(session.0.to_string()), Json(req))
            .await
            .expect("route succeeds");
    body
}

/// The `PreToolUse` payload the daemon's own `matcher: "*"` hook observes — the
/// ORIGINAL dispatch, before the guard rewrote it.
fn unisolated_hook_payload(tool_use_id: &str) -> Value {
    serde_json::json!({
        "cwd": "/repo",
        "tool": "Agent",
        "tool_use_id": tool_use_id,
        "input": {"subagent_type": "rust-engineer", "description": "go"},
    })
}

/// The Bash payload `tm hook` POSTs to this route for an ADR-0049
/// documents-only commit and for the ADR-0048 `git merge`/`git rebase` query.
/// It carries no `input`, so nothing about it is a dispatch.
fn commit_query(cwd: &str) -> SharedTreeDispatchRequest {
    SharedTreeDispatchRequest {
        payload: serde_json::json!({
            "cwd": cwd,
            "tool": "Bash",
            "input": {"command": "git commit -m 'docs: snapshot'"},
        }),
    }
}

/// The #6556 critic round 2. The occupancy split kept a type-reconciled record
/// counted for a dispatch — correctly — but the ADR-0049 documents-only commit
/// travels the SAME route, so every caller heard the occupant and the commit
/// #6556 exists to unblock stayed denied for the full
/// `RUNNING_STALE_AFTER_SECS` window. The unit-level assertion in
/// `daemon::state::tests` could not catch it: no consumer reads
/// `live_shared_tree_writers` through the route.
///
/// The input is #6556's own incident. A `rust-engineer` is dispatched into a
/// main checkout, its `agent_id` is never taught, and 80 minutes later its
/// `SubagentStop` reconciles the record by `agent_type`.
///
/// Fails against f69363eeb: the commit query answers one agent and
/// `docs_commit_deny_reason` fires.
#[tokio::test]
async fn the_commit_query_and_the_dispatch_query_disagree_on_one_record() {
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_lost"),
        DelegationStatus::Running,
    );
    // What `reconcile_by_agent_type` writes when the stop names this agent type
    // and only this record could be it.
    let id = state.delegations_for(session)[0].id;
    state.mutate_delegation(id, |d| {
        d.started_at = Some(chrono::Utc::now());
        d.status = DelegationStatus::Stale;
        d.stale_by_agent_type = true;
    });

    // The commit query: the reconciliation must be honoured, or #6556 is not
    // fixed at the surface it was filed against.
    let commit = call(&state, session, commit_query("/repo")).await;
    assert!(
        commit.agents.is_empty(),
        "a documents-only commit must not be denied by a type-reconciled record: {:?}",
        commit.agents
    );
    assert_eq!(commit.total, 0);
    assert!(!commit.claimed, "a Bash query never claims the tree");

    // The dispatch query: the tree is still occupied, because the
    // identification was by type and may name an agent still writing.
    let disp = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_new")),
    )
    .await;
    assert_eq!(
        disp.agents.len(),
        1,
        "a second file-mutating dispatch must still see the occupant: {:?}",
        disp.agents
    );
    assert_eq!(disp.agents[0].agent, "rust-engineer");
    assert!(
        !disp.claimed,
        "and must not be admitted onto that agent's HEAD"
    );
}

/// The control: with no type reconciliation in play the two questions agree, so
/// the split cannot be hiding a live writer from the commit guard. A genuinely
/// `Running` record denies both.
///
/// #6797: the writer is another SESSION's. It used to be inserted under the
/// asking session, which the commit query could not tell from a foreign writer;
/// now it can, and this test is about the foreign one — the caller's own record
/// is covered by `head_write_excludes_the_calling_sessions_own_delegation` and
/// `a_dispatch_still_sees_the_callers_own_occupant`.
#[tokio::test]
async fn a_live_writer_is_reported_to_both_questions() {
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        SessionId(uuid::Uuid::new_v4()),
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_live"),
        DelegationStatus::Running,
    );

    let commit = call(&state, session, commit_query("/repo")).await;
    assert_eq!(
        commit.agents.len(),
        1,
        "a live writer must still deny a documents-only commit (ADR-0049)"
    );

    let disp = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_new")),
    )
    .await;
    assert_eq!(disp.agents.len(), 1);
    assert!(!disp.claimed);
}

#[tokio::test]
async fn a_grant_and_the_tracker_converge_in_either_order() {
    // #5769, and the reason the fix is an upsert rather than a second call to
    // `observe`. TWO writers describe one dispatch: this route, carrying the
    // isolation `tm hook --pm-guard` granted, and the daemon's own `matcher: "*"`
    // tracker hook, carrying the ORIGINAL unisolated payload. They fire on the
    // same event and race, and `on_dispatch` returns early when the `tool_use_id`
    // already exists — so an `observe`-based grant would correct the record only
    // in the orders it happened to win. Both orders must converge on ONE isolated
    // record, or a HEAD move in this checkout is denied for the six hours of
    // `RUNNING_STALE_AFTER_SECS` on a writer that is not there.
    for guard_first in [true, false] {
        let (state, _dir, session) = hermetic();
        let original = unisolated_hook_payload("toolu_grant");
        let granted = dispatch(
            "/repo",
            "rust-engineer",
            Some("worktree"),
            Some("toolu_grant"),
        );

        if guard_first {
            let body = granted_call(&state, session, granted).await;
            assert!(body.claimed, "an empty checkout must be claimed");
            crate::daemon::services::delegation_tracker::observe(
                &state,
                session,
                HookEvent::PreToolUse,
                &original,
            );
        } else {
            crate::daemon::services::delegation_tracker::observe(
                &state,
                session,
                HookEvent::PreToolUse,
                &original,
            );
            // Causality: the tracker's record IS the phantom at this point.
            // Only an overwrite can clear it — `observe` would return early on
            // the `tool_use_id` it already wrote and change nothing.
            assert_eq!(state.delegations_for(session)[0].isolation, None);
            let body = granted_call(&state, session, granted).await;
            assert!(
                body.claimed,
                "the caller's own record is excluded, so the checkout is still free"
            );
        }

        let records = state.delegations_for(session);
        assert_eq!(
            records.len(),
            1,
            "guard_first={guard_first}: one dispatch must leave one record, got {records:?}"
        );
        assert_eq!(
            records[0].isolation.as_deref(),
            Some("worktree"),
            "guard_first={guard_first}: the granted isolation must survive both orders"
        );
        // The property the whole fix exists for: a later `git merge` in this
        // checkout asks exactly this question, and a non-empty answer denies it.
        assert!(
            state
                .live_shared_tree_writers(&PathBuf::from("/repo"), None)
                .is_empty(),
            "guard_first={guard_first}: a granted writer must not be named as writing here"
        );
    }
}

#[test]
fn a_grant_and_the_tracker_race_without_losing_the_isolation() {
    // What `dispatch_record` exists for, and the only test that can fail if it
    // is deleted. The sibling above drives both writers sequentially on one
    // thread: it proves the two ARRIVAL ORDERS converge, and proves nothing
    // about the locked window between a writer's find and its insert.
    //
    // Two ways the unguarded interleaving loses. If `observe` inserts between
    // `record_granted_isolation`'s find and `on_dispatch_locked`'s own find,
    // that second find hits and returns early — one record, isolation `None`,
    // a silent lost update. If neither find sees the other, both insert — two
    // records for one dispatch, and the unisolated one decides later denies.
    // Both assertions below are needed; neither alone catches both shapes.
    const ROUNDS: usize = 300;
    let rt = tokio::runtime::Builder::new_current_thread()
        .build()
        .expect("runtime");
    for round in 0..ROUNDS {
        let (state, _dir, session) = hermetic();
        let payload = unisolated_hook_payload("toolu_race");
        let barrier = std::sync::Barrier::new(2);
        std::thread::scope(|scope| {
            scope.spawn(|| {
                barrier.wait();
                crate::daemon::services::delegation_tracker::observe(
                    &state,
                    session,
                    HookEvent::PreToolUse,
                    &payload,
                );
            });
            barrier.wait();
            rt.block_on(granted_call(
                &state,
                session,
                dispatch(
                    "/repo",
                    "rust-engineer",
                    Some("worktree"),
                    Some("toolu_race"),
                ),
            ));
        });

        let records = state.delegations_for(session);
        assert_eq!(
            records.len(),
            1,
            "round {round}: one dispatch must leave one record, got {records:?}"
        );
        assert_eq!(
            records[0].isolation.as_deref(),
            Some("worktree"),
            "round {round}: the grant must not be lost to the tracker's insert"
        );
    }
}

#[tokio::test]
async fn granted_worktree_route_records_nothing_without_isolation_or_a_tool_use_id() {
    // Both keys are load-bearing. Without `tool_use_id` the record could never
    // be found again, so writing one would leave a SECOND record beside the
    // tracker's — the phantom duplicated rather than removed. Without an
    // isolating mode there is no grant to record, and this route must not become
    // a way to claim a directory for an ordinary unisolated dispatch.
    for req in [
        dispatch("/repo", "rust-engineer", Some("worktree"), None),
        dispatch("/repo", "rust-engineer", None, Some("toolu_x")),
        dispatch("/repo", "rust-engineer", Some("nonsense"), Some("toolu_x")),
    ] {
        let (state, _dir, session) = hermetic();
        let body = granted_call(&state, session, req).await;
        assert!(!body.claimed);
        assert!(state.delegations_for(session).is_empty());
    }
}

#[tokio::test]
async fn record_granted_isolation_refuses_a_non_separating_mode() {
    // The rule lives in the WRITER, not only in the route's eligibility test.
    // This function exists to erase a record that names a writer as sharing the
    // checkout, so a caller handing it `isolation: "inline"` would write that
    // phantom over a correct record — the exact inverse of its purpose.
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo",
        Some("worktree"),
        Some("toolu_x"),
        DelegationStatus::Running,
    );
    for mode in ["inline", "Worktree", "worktrees", ""] {
        let payload = serde_json::json!({
            "cwd": "/repo",
            "tool": "Agent",
            "tool_use_id": "toolu_x",
            "input": {"subagent_type": "rust-engineer", "isolation": mode},
        });
        assert!(
            !crate::daemon::services::delegation_tracker::record_granted_isolation(
                &state, session, &payload
            ),
            "{mode:?} does not separate the working tree and must not be recorded"
        );
    }
    assert_eq!(
        state.delegations_for(session)[0].isolation.as_deref(),
        Some("worktree"),
        "a correct record must survive every refused call"
    );
}

#[tokio::test]
async fn granted_worktree_route_reports_a_live_writer_without_claiming() {
    // The deny arm the reorder restored (#5769 finding 2): the grant is emitted
    // only after this answer comes back empty, so a sibling already holding the
    // checkout denies the dispatch instead of silently sharing it — which is
    // what happened while the grant returned before the concurrency check.
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "python-engineer",
        "/repo",
        None,
        Some("toolu_first"),
        DelegationStatus::Running,
    );
    let body = granted_call(
        &state,
        session,
        dispatch(
            "/repo",
            "rust-engineer",
            Some("worktree"),
            Some("toolu_second"),
        ),
    )
    .await;
    assert_eq!(body.total, 1);
    assert_eq!(body.agents[0].agent, "python-engineer");
    assert!(!body.claimed, "a denied dispatch must claim nothing");
    // #7487: the deny leaves ONE Cancelled tombstone beside the occupant, never
    // a live record and never a second tombstone. The total is pinned as well as
    // the live count — an implementation that wrote a fresh Cancelled record
    // instead of cancelling the one it found passes the live filter alone.
    let all = state.delegations_for(session);
    assert_eq!(all.len(), 2, "one occupant plus one tombstone: {all:?}");
    let denied = all
        .iter()
        .find(|d| d.tool_use_id.as_deref() == Some("toolu_second"))
        .expect("the denied dispatch has a record");
    assert_eq!(denied.status, DelegationStatus::Cancelled);
    let live: Vec<_> = all.iter().filter(|d| d.status.is_live()).collect();
    assert_eq!(live.len(), 1);
    assert_eq!(live[0].agent, "python-engineer");
}

#[tokio::test]
async fn shared_tree_dispatch_route_reports_live_unisolated_writers() {
    let (state, _dir, session) = hermetic();
    // The one that matters: a live engineer in the shared tree with no
    // isolation. This is the record that makes a second dispatch dangerous.
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_A"),
        DelegationStatus::Running,
    );
    // Each of the following must be filtered out, and each for its own reason.
    // A worktree-isolated sibling is exactly what the guard wants the PM to do.
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo",
        Some("worktree"),
        Some("toolu_B"),
        DelegationStatus::Running,
    );
    // A read-only agent shares the tree harmlessly.
    insert(
        &state,
        session,
        "research",
        "/repo",
        None,
        Some("toolu_C"),
        DelegationStatus::Running,
    );
    // A finished engineer is not in flight.
    insert(
        &state,
        session,
        "python-engineer",
        "/repo",
        None,
        Some("toolu_D"),
        DelegationStatus::Completed,
    );
    // A `Stale` record is one tracking gave up on — it must not block a
    // dispatch for the rest of its retention window.
    insert(
        &state,
        session,
        "python-engineer",
        "/repo",
        None,
        Some("toolu_E"),
        DelegationStatus::Stale,
    );
    // A different directory is a different race.
    insert(
        &state,
        session,
        "python-engineer",
        "/elsewhere",
        None,
        Some("toolu_F"),
        DelegationStatus::Running,
    );
    // Another session's child in THIS directory now counts (ADR-0048). This
    // line used to assert the opposite — "not this session's problem" — and
    // that assumption is what made the guard blind to the reported incident:
    // the writers sharing one checkout each belonged to a different session.
    // Dedicated coverage is `shared_tree_writers_span_sessions_in_one_checkout`.
    insert(
        &state,
        SessionId(uuid::Uuid::new_v4()),
        "python-engineer",
        "/repo",
        None,
        Some("toolu_G"),
        DelegationStatus::Running,
    );

    let body = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_NEW")),
    )
    .await;

    assert_eq!(
        body.total, 2,
        "the live unisolated engineers in this directory, whichever session dispatched them"
    );
    assert_eq!(body.agents.len(), 2, "{:?}", body.agents);
    assert_eq!(body.agents[0].agent, "python-engineer");
    assert_eq!(body.agents[1].agent, "rust-engineer");
    assert!(!body.claimed, "a denied dispatch must not claim the tree");
}

#[tokio::test]
async fn shared_tree_dispatch_route_excludes_the_callers_own_dispatch() {
    // Causality: the daemon's `matcher: "*"` hook and `tm hook --pm-guard`
    // race on the same PreToolUse. If the tracker wins, the asking dispatch is
    // already recorded — and without this exclusion the FIRST dispatch of a
    // session would find itself and be denied.
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_SELF"),
        DelegationStatus::Running,
    );

    let body = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_SELF")),
    )
    .await;

    assert_eq!(body.total, 0, "the caller must not find itself");
    // Redelivery of the same dispatch must not add a second record either —
    // the tracker's observer is idempotent on `tool_use_id`.
    assert_eq!(state.delegations_for(session).len(), 1);
}

#[tokio::test]
async fn shared_tree_dispatch_route_counts_concurrent_same_agent_dispatches() {
    let (state, _dir, session) = hermetic();
    for id in ["toolu_1", "toolu_2"] {
        insert(
            &state,
            session,
            "rust-engineer",
            "/repo",
            None,
            Some(id),
            DelegationStatus::Running,
        );
    }

    let body = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_3")),
    )
    .await;

    assert_eq!(body.total, 2);
    assert_eq!(body.agents.len(), 1, "one row per distinct name");
    assert_eq!(body.agents[0].count, 2);
}

#[tokio::test]
async fn shared_tree_dispatch_route_is_empty_for_an_unknown_session() {
    // An unknown session has no delegations. Answering 404 would read to the
    // guard as an error rather than as "nobody else is here".
    let (state, _dir, _session) = hermetic();
    let Json(body) = shared_tree_dispatch_route(
        State(state),
        Path(uuid::Uuid::new_v4().to_string()),
        Json(dispatch("/repo", "rust-engineer", None, Some("toolu_X"))),
    )
    .await
    .expect("route succeeds");
    assert_eq!(body.total, 0);
}

#[tokio::test]
async fn shared_tree_dispatch_route_rejects_a_malformed_session_id() {
    let (state, _dir, _session) = hermetic();
    let err = shared_tree_dispatch_route(
        State(state),
        Path("not-a-uuid".to_string()),
        Json(dispatch("/repo", "rust-engineer", None, Some("toolu_X"))),
    )
    .await
    .expect_err("a malformed id is a client error");
    assert!(matches!(err, DaemonError::InvalidRequest(_)));
}

// ---------------------------------------------------------------------------
// #5324 — the claim
// ---------------------------------------------------------------------------

#[tokio::test]
async fn shared_tree_dispatch_route_denies_the_second_claim() {
    // THE regression case. Two dispatches into one directory, neither recorded
    // by anything else. Pre-#5324 this route only answered, so BOTH calls saw an
    // empty set and both were admitted. The claim taken by the first is what
    // the second now finds.
    let (state, _dir, session) = hermetic();

    let first = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_A")),
    )
    .await;
    assert_eq!(first.total, 0, "the first dispatch must be admitted");
    assert!(first.claimed, "and it must take the directory");

    let second = call(
        &state,
        session,
        dispatch("/repo", "python-engineer", None, Some("toolu_B")),
    )
    .await;
    assert_eq!(second.total, 1, "the second must find the first");
    assert_eq!(second.agents[0].agent, "rust-engineer");
    assert!(!second.claimed);
}

#[tokio::test]
async fn shared_tree_dispatch_route_reserves_the_tree_on_an_empty_answer() {
    // The claim is not a new kind of state: it is the delegation record the
    // tracker's own PreToolUse observer writes, carrying the same correlation
    // key, directory, and isolation — so it is released by the same
    // `SubagentStop` and swept by the same staleness sweep.
    let (state, _dir, session) = hermetic();
    call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_A")),
    )
    .await;

    let records = state.delegations_for(session);
    assert_eq!(records.len(), 1);
    assert_eq!(records[0].agent, "rust-engineer");
    assert_eq!(records[0].tool_use_id.as_deref(), Some("toolu_A"));
    assert_eq!(
        records[0].cwd.as_deref(),
        Some(PathBuf::from("/repo").as_path())
    );
    assert_eq!(records[0].status, DelegationStatus::Running);
}

#[tokio::test]
async fn shared_tree_dispatch_route_does_not_reserve_when_it_denies() {
    // A denied dispatch never runs, so recording it would occupy the directory
    // for a subagent that does not exist — a false-deny generator.
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_A"),
        DelegationStatus::Running,
    );

    let body = call(
        &state,
        session,
        dispatch("/repo", "python-engineer", None, Some("toolu_B")),
    )
    .await;

    assert!(!body.claimed);
    // #7487: the guarantee is unchanged — a denied dispatch must never OCCUPY
    // the directory — but its shape is. The deny now leaves a TERMINAL tombstone
    // on `toolu_B` instead of nothing at all, because the tracker's
    // `matcher: "*"` hook POSTs the same dispatch independently and can land
    // AFTER this call; an absent record is exactly what let that late
    // observation write a live claim for an agent that never ran. A terminal
    // record occupies nothing and makes `on_dispatch` a no-op on that id.
    assert_eq!(
        state.shared_tree_occupants(std::path::Path::new("/repo"), None),
        vec!["rust-engineer".to_string()],
        "the denied dispatch must not occupy the directory"
    );
    // The record pin the occupancy assertion cannot make: exactly ONE tombstone
    // is added, so a deny can never multiply records for one `tool_use_id`.
    let records = state.delegations_for(session);
    assert_eq!(
        records.len(),
        2,
        "the live writer plus exactly one tombstone: {records:?}"
    );
    let denied = records
        .into_iter()
        .find(|d| d.tool_use_id.as_deref() == Some("toolu_B"))
        .expect("the deny records a tombstone so a late observation cannot revive it");
    assert_eq!(
        denied.status,
        DelegationStatus::Cancelled,
        "and that tombstone must be terminal"
    );
}

#[tokio::test]
async fn shared_tree_dispatch_route_does_not_reserve_a_read_only_agent() {
    // The daemon re-derives eligibility rather than trusting the caller: a
    // research/review dispatch, or one that declared isolation, must never
    // occupy a directory even if something asked it to.
    for (agent, isolation) in [
        ("research", None),
        ("code-critic", None),
        ("rust-engineer", Some("worktree")),
        ("rust-engineer", Some("remote")),
        ("some-project-agent", None),
    ] {
        let (state, _dir, session) = hermetic();
        let body = call(
            &state,
            session,
            dispatch("/repo", agent, isolation, Some("toolu_A")),
        )
        .await;
        assert!(
            !body.claimed,
            "{agent}/{isolation:?} must not claim the tree"
        );
        assert!(state.delegations_for(session).is_empty());
    }
}

#[tokio::test]
async fn shared_tree_dispatch_route_does_not_reserve_a_non_dispatch_tool() {
    // The recording path is the tracker's, which acts only on a dispatch tool.
    // Reporting a claim it did not take would be a lie the response tells.
    let (state, _dir, session) = hermetic();
    let mut req = dispatch("/repo", "rust-engineer", None, Some("toolu_A"));
    req.payload["tool"] = Value::String("Read".into());
    let body = call(&state, session, req).await;
    assert!(!body.claimed);
    assert!(state.delegations_for(session).is_empty());
}

#[tokio::test]
async fn shared_tree_dispatch_route_answers_a_bash_query_without_claiming() {
    // ADR-0048 decision 10: the HEAD-moving Bash rule reads this route through
    // `pm_guard_dispatch::live_shared_tree_writers`, which sends the Bash call's
    // own payload — `tool: "Bash"` and an `input` projected to nothing. Two
    // halves of that contract are pinned here, because the rule depends on both.
    // It must ANSWER: a `git merge` beside a live writer is the deny this exists
    // for. And it must claim NOTHING: a pull is not a dispatch, and a claim it
    // took would occupy a directory no `SubagentStop` will ever release.
    //
    // #6797: the writer is another SESSION's, which is what "a live writer" meant
    // here all along — a HEAD write no longer hears its own session's agents.
    let (state, _dir, session) = hermetic();
    let stranger = SessionId(uuid::Uuid::new_v4());
    insert(
        &state,
        stranger,
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_A"),
        DelegationStatus::Running,
    );
    let req = SharedTreeDispatchRequest {
        payload: serde_json::json!({
            "cwd": "/repo",
            "tool": "Bash",
            "input": {},
            "tool_use_id": "toolu_pull",
        }),
    };
    let body = call(&state, session, req).await;
    assert_eq!(body.total, 1, "the live writer must be reported");
    assert_eq!(body.agents[0].agent, "rust-engineer");
    assert!(!body.claimed, "a Bash query must never claim the tree");
    assert!(
        state.delegations_for(session).is_empty(),
        "a Bash query must add no record for the session that asked"
    );
    assert_eq!(
        state.delegations_for(stranger).len(),
        1,
        "and must leave the writer's own record exactly as it found it"
    );
}

#[tokio::test]
async fn shared_tree_dispatch_route_is_empty_without_a_cwd() {
    // Declared fail-open branch: with no directory in the payload there is
    // nothing to compare against and nothing to claim. Seeding a live writer
    // makes the assertion causal — only the missing-cwd short-circuit can keep
    // the answer empty.
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_A"),
        DelegationStatus::Running,
    );
    let mut req = dispatch("/repo", "rust-engineer", None, Some("toolu_B"));
    req.payload["cwd"] = Value::String(String::new());

    let body = call(&state, session, req).await;
    assert_eq!(body.total, 0);
    assert!(!body.claimed);
    assert_eq!(state.delegations_for(session).len(), 1);
}

#[tokio::test]
async fn shared_tree_writers_span_sessions_in_one_checkout() {
    // THE REGRESSION (ADR-0048). The reported incident, reduced: three sessions
    // standing in one `mcp-services` checkout, each dispatching a writer into
    // it. Every guard saw an empty answer and admitted its writer, because the
    // answer was filtered to the ASKING session's own delegations before any
    // other test ran — so the writers were invisible to each other by
    // construction, not by timing. Downstream that produced branches switching
    // under each other and a commit landing on a workstream it did not belong
    // to.
    //
    // Session B asks about the directory session A's writer is already in.
    // With the session filter this answered 0 and claimed the tree a second
    // time; the hazard is a shared git HEAD, which belongs to the DIRECTORY and
    // knows nothing about session ids.
    let (state, _dir, session_a) = hermetic();
    let session_b = SessionId(uuid::Uuid::new_v4());
    let session_c = SessionId(uuid::Uuid::new_v4());

    insert(
        &state,
        session_a,
        "rust-engineer",
        "/repo/mcp-services",
        None,
        Some("toolu_A"),
        DelegationStatus::Running,
    );

    let body = call(
        &state,
        session_b,
        dispatch(
            "/repo/mcp-services",
            "python-engineer",
            None,
            Some("toolu_B"),
        ),
    )
    .await;
    assert_eq!(
        body.total, 1,
        "session B must see session A's writer in the same checkout"
    );
    assert_eq!(body.agents[0].agent, "rust-engineer");
    assert!(
        !body.claimed,
        "a denied dispatch must not also occupy the directory"
    );

    // A third session gets the same answer, naming every writer it would be
    // joining rather than only the ones its own session dispatched.
    insert(
        &state,
        session_b,
        "documentation",
        "/repo/mcp-services",
        None,
        Some("toolu_B2"),
        DelegationStatus::Running,
    );
    let body = call(
        &state,
        session_c,
        dispatch("/repo/mcp-services", "qa", None, Some("toolu_C")),
    )
    .await;
    assert_eq!(body.total, 2, "every live writer in the directory counts");

    // The widening must not reach across DIRECTORIES — that would deny every
    // dispatch on the machine as soon as one agent was running anywhere. A
    // worktree is a different directory and stays free, which is what makes the
    // remedy the deny offers actually work.
    let body = call(
        &state,
        session_c,
        dispatch(
            "/repo/mcp-services/.claude/worktrees/w1",
            "rust-engineer",
            None,
            Some("toolu_D"),
        ),
    )
    .await;
    assert_eq!(
        body.total, 0,
        "a different directory is a different git HEAD and must stay admitted"
    );
    assert!(body.claimed, "the first writer in its own tree claims it");
}

#[tokio::test]
async fn shared_tree_dispatch_route_claim_is_idempotent_on_redelivery() {
    // A redelivered hook must not deny the dispatch it already admitted, and
    // must not double-record it. Both follow from excluding the caller's own
    // `tool_use_id` and from the observer's own idempotence.
    let (state, _dir, session) = hermetic();
    for _ in 0..3 {
        let body = call(
            &state,
            session,
            dispatch("/repo", "rust-engineer", None, Some("toolu_A")),
        )
        .await;
        assert_eq!(body.total, 0, "a redelivery must stay admitted");
    }
    assert_eq!(state.delegations_for(session).len(), 1);
}

// ── #6797: a session's own subagent is not a foreign live writer ────────────

/// The reported incident. A session dispatches an agent into its main checkout,
/// the agent's record stays `Running` because no `SubagentStop` closed it, and
/// the session's next ADR-0049 documents-only commit is refused — naming an agent
/// of the very session that is committing.
///
/// Live reproduction 2026-09-04 against the running daemon: the commit query for
/// `/Users/masa/trusty-mpm-projects/bobmatnyc/trusty-tools` answered
/// `{"agents":[{"agent":"qa","count":2},{"agent":"rust-engineer","count":1}]}`,
/// and every one of those records carried the CALLING session's id.
#[tokio::test]
async fn head_write_excludes_the_calling_sessions_own_delegation() {
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_own"),
        DelegationStatus::Running,
    );

    let commit = call(&state, session, commit_query("/repo")).await;
    assert!(
        commit.agents.is_empty(),
        "a session's own agent must not be reported to it as a foreign live writer: {:?}",
        commit.agents
    );
    assert_eq!(commit.total, 0);
    assert!(!commit.claimed, "a Bash query never claims the tree");
    assert!(
        !commit.tree_holders,
        "no marker sent, so none echoed (#8161)"
    );
}

/// #8161: the tree-holders marker lifts #6797's own-session exclusion, because a
/// consolidating `reset --keep` would clobber the asking PM's own live agent.
#[tokio::test]
async fn shared_tree_route_tree_holders_counts_the_callers_own_agent() {
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo/.claude/worktrees/agent-parked",
        None,
        Some("toolu_own"),
        DelegationStatus::Running,
    );
    let mut query = commit_query("/repo/.claude/worktrees/agent-parked");
    query.payload[crate::daemon::delegation_routes::TREE_HOLDERS_MARKER] = Value::Bool(true);

    let answer = call(&state, session, query).await;
    assert_eq!(answer.total, 1, "{:?}", answer.agents);
    assert!(
        answer.tree_holders,
        "the answer must echo the marker it honoured"
    );
    assert!(!answer.claimed, "a Bash query never claims the tree");
}

/// The fail-closed half, and the reason the exclusion is scoped to the caller
/// rather than applied to every record. ADR-0048 decision 10's hazard is ANOTHER
/// session's uncommitted work, so another session's record must still deny.
#[tokio::test]
async fn head_write_still_reports_another_sessions_delegation() {
    let (state, _dir, caller) = hermetic();
    let stranger = SessionId(uuid::Uuid::new_v4());
    insert(
        &state,
        stranger,
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_other"),
        DelegationStatus::Running,
    );

    let commit = call(&state, caller, commit_query("/repo")).await;
    assert_eq!(
        commit.agents.len(),
        1,
        "another session's live writer must still deny: {:?}",
        commit.agents
    );
    assert_eq!(commit.agents[0].agent, "rust-engineer");
    assert_eq!(commit.total, 1);
}

/// The exclusion must not weaken the ADMISSION answer. Whether two unisolated
/// file-mutating agents may share one main checkout cannot vary with who is
/// asking, so the dispatch query still sees the caller's own occupant and denies
/// the second dispatch — ADR-0048 decision 3 and #4480 unchanged.
#[tokio::test]
async fn a_dispatch_still_sees_the_callers_own_occupant() {
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_own"),
        DelegationStatus::Running,
    );

    let disp = call(
        &state,
        session,
        dispatch("/repo", "qa", None, Some("toolu_second")),
    )
    .await;
    assert_eq!(
        disp.total, 1,
        "a second unisolated writer must still be denied by the caller's own occupant: {:?}",
        disp.agents
    );
    assert!(
        !disp.claimed,
        "and the tree must not be claimed for a dispatch that is denied"
    );
}

/// A commit and a dispatch from the same session, over the same record, must
/// disagree — which is the whole reason the caller identity rides on
/// `SharedTreeQuestion::HeadWrite` rather than sitting beside the question.
#[tokio::test]
async fn the_two_questions_disagree_about_the_callers_own_record() {
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "rust-engineer",
        "/repo",
        None,
        Some("toolu_own"),
        DelegationStatus::Running,
    );

    let commit = call(&state, session, commit_query("/repo")).await;
    let disp = call(
        &state,
        session,
        dispatch("/repo", "qa", None, Some("toolu_second")),
    )
    .await;
    assert_eq!(commit.total, 0, "the commit hears nothing of its own agent");
    assert_eq!(disp.total, 1, "the dispatch still hears the occupant");
}

// ── #7487: a claim outlives neither a TaskStop nor its own denial ────────────

/// Teach the tracker the `agentId` a dispatch's `PostToolUse` carries (#7487).
fn launch(state: &Arc<DaemonState>, session: SessionId, tool_use_id: &str, agent_id: &str) {
    crate::daemon::services::delegation_tracker::observe(
        state,
        session,
        crate::core::hook::HookEvent::PostToolUse,
        &serde_json::json!({
            "cwd": "/repo",
            "tool": "Agent",
            "tool_use_id": tool_use_id,
            "tool_response": {"isAsync": true, "agentId": agent_id},
        }),
    );
}

/// The `PostToolUse` a `TaskStop` emits, as `tm hook` forwards it (#7487).
///
/// `outcome` is the `tool_response` the hook projection carries — the REAL wire
/// discrimination between a stop that took and one that did not (#7511 review).
/// `HookEvent::PostToolUseFailure` is deliberately not used: no hook block `tm`
/// writes registers that name, so a test driving it would prove the branch and
/// not the production path.
fn task_stop(state: &Arc<DaemonState>, session: SessionId, task_id: &str, outcome: Option<Value>) {
    let mut payload = serde_json::json!({
        "cwd": "/repo",
        "tool": "TaskStop",
        "tool_use_id": "toolu_stop",
        "input": {"task_id": task_id},
    });
    if let Some(response) = outcome {
        payload["tool_response"] = response;
    }
    crate::daemon::services::delegation_tracker::observe(
        state,
        session,
        crate::core::hook::HookEvent::PostToolUse,
        &payload,
    );
}

/// What `tm hook` forwards for a stop that took: `is_error` present and false.
fn stop_succeeded() -> Option<Value> {
    Some(serde_json::json!({"is_error": false}))
}

/// What `tm hook` forwards for a stop that did NOT take.
fn stop_failed() -> Option<Value> {
    Some(serde_json::json!({"is_error": true}))
}

/// 🔴 REGRESSION (#7487a): a `TaskStop`-cancelled agent releases the tree.
///
/// Why: a stopped subagent emits no `SubagentStop`, so nothing terminalized its
/// record and `shared_tree_occupants` kept naming it for the six hours of
/// `RUNNING_STALE_AFTER_SECS`. Fails on 4cc327dbc, where the third dispatch is
/// still refused by an agent that was stopped two minutes earlier.
#[tokio::test]
async fn a_task_stop_releases_the_stopped_agents_claim() {
    let (state, _dir, session) = hermetic();

    let first = call(
        &state,
        session,
        dispatch("/repo", "qa", None, Some("toolu_qa")),
    )
    .await;
    assert!(first.claimed, "the first dispatch takes the tree");
    launch(&state, session, "toolu_qa", "agent-qa-1");

    // Control: while `qa` is running, a second unisolated writer is refused.
    let while_running = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_a")),
    )
    .await;
    assert_eq!(while_running.total, 1, "a live claim must still refuse");
    assert!(!while_running.claimed);

    task_stop(&state, session, "agent-qa-1", stop_succeeded());

    let after_stop = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_b")),
    )
    .await;
    assert_eq!(
        after_stop.total, 0,
        "a stopped agent holds nothing; occupants were {:?}",
        after_stop.agents
    );
    assert!(after_stop.claimed, "the tree must be claimable again");
}

/// 🔴 A stop that did NOT take leaves the agent running, so it releases nothing.
///
/// The discrimination is the forwarded `tool_response`, which is the only thing
/// on the wire that says whether the stop succeeded (#7511 review, MEDIUM 2+3).
/// Fails on c7eb75d5b, which released on the `PostToolUse` event alone.
#[tokio::test]
async fn a_failed_task_stop_releases_nothing() {
    let (state, _dir, session) = hermetic();
    call(
        &state,
        session,
        dispatch("/repo", "qa", None, Some("toolu_qa")),
    )
    .await;
    launch(&state, session, "toolu_qa", "agent-qa-1");

    task_stop(&state, session, "agent-qa-1", stop_failed());

    let after = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_a")),
    )
    .await;
    assert_eq!(after.total, 1, "a failed stop must not release the tree");
    assert!(!after.claimed);
}

/// 🔴 The documented fail-OPEN branch: a stop whose response says nothing still
/// releases.
///
/// Why: `TaskStop`'s response shape is pinned by no contract this repo owns, and
/// `tm hook` forwards only `TOOL_RESPONSE_KEYS` — so a plain-string or
/// `is_error`-less answer arrives as no response at all. Requiring positive
/// proof of success would release nothing, ever, reinstating #7487 silently.
/// This pins the choice so a later tightening has to argue with it rather than
/// make it by accident.
#[tokio::test]
async fn a_task_stop_with_no_response_still_releases() {
    let (state, _dir, session) = hermetic();
    call(
        &state,
        session,
        dispatch("/repo", "qa", None, Some("toolu_qa")),
    )
    .await;
    launch(&state, session, "toolu_qa", "agent-qa-1");

    task_stop(&state, session, "agent-qa-1", None);

    let after = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_a")),
    )
    .await;
    assert_eq!(
        after.total, 0,
        "a response that cannot say either way must not re-lock the tree"
    );
    assert!(after.claimed);
}

/// 🔴 A `task_id` no delegation carries terminalizes nothing.
#[tokio::test]
async fn a_task_stop_naming_an_unknown_id_terminalizes_nothing() {
    let (state, _dir, session) = hermetic();
    call(
        &state,
        session,
        dispatch("/repo", "qa", None, Some("toolu_qa")),
    )
    .await;
    launch(&state, session, "toolu_qa", "agent-qa-1");

    task_stop(&state, session, "some-background-shell", stop_succeeded());

    let after = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_a")),
    )
    .await;
    assert_eq!(
        after.total, 1,
        "a stop naming something else must close nothing — a 'most recent live \
         record' guess is what the tracker forbids"
    );
}

/// 🔴 REGRESSION (#7487b): a dispatch the guard denies records no claim, when
/// the tracker's own hook lands AFTER the deny.
///
/// Why: two `PreToolUse` hooks see one dispatch — `tm hook --pm-guard` decides
/// and the `matcher: "*"` tracker records, independently and in either order —
/// so a denied dispatch left a live record carrying its `cwd` and became the
/// next dispatch's collision. The reported incident's second refusal read
/// "qa, rust-engineer", naming the very dispatch the guard had just refused.
/// Fails on 4cc327dbc, where the third call hears two occupants.
#[tokio::test]
async fn a_late_observation_of_a_denied_dispatch_records_nothing() {
    let (state, _dir, session) = hermetic();
    call(
        &state,
        session,
        dispatch("/repo", "qa", None, Some("toolu_qa")),
    )
    .await;
    launch(&state, session, "toolu_qa", "agent-qa-1");
    call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_denied")),
    )
    .await;

    // The `matcher: "*"` hook's POST, arriving after the guard already denied.
    crate::daemon::services::delegation_tracker::observe(
        &state,
        session,
        crate::core::hook::HookEvent::PreToolUse,
        &unisolated_hook_payload("toolu_denied"),
    );

    let next = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_next")),
    )
    .await;
    assert_eq!(
        next.total, 1,
        "a late observation must not resurrect a refused dispatch's claim. Got {:?}",
        next.agents
    );
    let names: Vec<&str> = next.agents.iter().map(|a| a.agent.as_str()).collect();
    assert_eq!(
        names,
        vec!["qa"],
        "a denial must not manufacture an occupant"
    );
}

/// 🔴 The mirror order: the tracker's hook lands FIRST, and the deny must
/// cancel the record it already wrote. Fails on 4cc327dbc for the same reason.
#[tokio::test]
async fn a_denied_dispatch_cancels_a_record_the_tracker_already_wrote() {
    let (state, _dir, session) = hermetic();
    call(
        &state,
        session,
        dispatch("/repo", "qa", None, Some("toolu_qa")),
    )
    .await;
    launch(&state, session, "toolu_qa", "agent-qa-1");

    // The `matcher: "*"` hook wins the race and records the dispatch first.
    crate::daemon::services::delegation_tracker::observe(
        &state,
        session,
        crate::core::hook::HookEvent::PreToolUse,
        &unisolated_hook_payload("toolu_denied"),
    );
    call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_denied")),
    )
    .await;

    let next = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_next")),
    )
    .await;
    assert_eq!(
        next.total, 1,
        "the deny must cancel the record the tracker had already written. Got {:?}",
        next.agents
    );
}

/// The distinct agent names a route answer reports.
fn writer_names(body: &SharedTreeWritersResponse) -> Vec<&str> {
    body.agents.iter().map(|a| a.agent.as_str()).collect()
}

/// 🔴 REGRESSION (#7487, recurrence 2026-09-13): a dispatch the ADR-0048 GRANT
/// path denies records no claim either.
///
/// Why: the 2026-09-11 fix released a denied dispatch on the shared-tree route
/// alone. From a main checkout the guard asks the granted-worktree route, whose
/// deny released nothing, so the tracker's `matcher: "*"` record of the ORIGINAL
/// unisolated payload became the next dispatch's occupant. The reported refusals
/// read "rust-engineer, version-control" and then "documentation, rust-engineer,
/// version-control", each naming a dispatch the guard had itself just refused.
/// Both hook orders are driven, as in
/// `a_grant_and_the_tracker_converge_in_either_order`.
/// Fails on 777153e25: the second grant hears `rust-engineer` beside
/// `version-control`.
#[tokio::test]
async fn a_dispatch_the_grant_path_denies_records_no_claim_7487() {
    for tracker_first in [false, true] {
        let (state, _dir, session) = hermetic();
        // ADR-0056: version-control stays in the main checkout, unisolated.
        insert(
            &state,
            session,
            "version-control",
            "/repo",
            None,
            Some("toolu_vc"),
            DelegationStatus::Running,
        );
        let tracker_hook = || {
            crate::daemon::services::delegation_tracker::observe(
                &state,
                session,
                HookEvent::PreToolUse,
                &unisolated_hook_payload("toolu_denied"),
            );
        };

        if tracker_first {
            tracker_hook();
        }
        let denied = granted_call(
            &state,
            session,
            dispatch(
                "/repo",
                "rust-engineer",
                Some("worktree"),
                Some("toolu_denied"),
            ),
        )
        .await;
        assert!(!denied.claimed, "tracker_first={tracker_first}");
        assert_eq!(writer_names(&denied), vec!["version-control"]);
        if !tracker_first {
            tracker_hook();
        }

        assert_eq!(
            state.shared_tree_occupants(std::path::Path::new("/repo"), None),
            vec!["version-control".to_string()],
            "tracker_first={tracker_first}: a refused dispatch must not occupy the checkout"
        );
        let next = granted_call(
            &state,
            session,
            dispatch(
                "/repo",
                "documentation",
                Some("worktree"),
                Some("toolu_doc"),
            ),
        )
        .await;
        assert_eq!(
            writer_names(&next),
            vec!["version-control"],
            "tracker_first={tracker_first}: the next deny must name only the real occupant"
        );
    }
}

/// 🔴 Fail-open pin (#7487): releasing a denied grant never releases the
/// occupant that caused the deny. That agent is still writing, and dropping its
/// record would admit a second writer onto its HEAD.
#[tokio::test]
async fn a_grant_deny_keeps_the_running_occupant_counted_7487() {
    let (state, _dir, session) = hermetic();
    insert(
        &state,
        session,
        "version-control",
        "/repo",
        None,
        Some("toolu_vc"),
        DelegationStatus::Running,
    );
    launch(&state, session, "toolu_vc", "agent-vc-1");

    granted_call(
        &state,
        session,
        dispatch(
            "/repo",
            "rust-engineer",
            Some("worktree"),
            Some("toolu_denied"),
        ),
    )
    .await;

    let unisolated = call(
        &state,
        session,
        dispatch("/repo", "rust-engineer", None, Some("toolu_next")),
    )
    .await;
    assert_eq!(writer_names(&unisolated), vec!["version-control"]);
    assert!(!unisolated.claimed, "the live occupant must still refuse");
    // #6797: a HEAD write never hears its own session's agents, so ask as another.
    let other_session = SessionId(uuid::Uuid::new_v4());
    let head_write = call(&state, other_session, commit_query("/repo")).await;
    assert_eq!(
        head_write.total, 1,
        "a HEAD write in another session's view must still see the live writer"
    );
}

/// 🔴 REGRESSION (#7487 critic round, reachable once #6556's retry lands): a
/// deny and then an ADMISSION on the same `tool_use_id` must leave a live
/// record, not the deny's tombstone wearing an `isolation`.
///
/// Why: the guard's ADR-0048 rewrite keeps the dispatch's `tool_use_id`, so the
/// unisolated form that was denied and the granted form that runs are the same
/// id. `record_granted_isolation` found the `Cancelled` tombstone and set only
/// `isolation`, so a running agent was described as Ended — its builder slot
/// free and, worse, its worktree reclaimable under it, which is #5661's shape.
/// Fails at 9727aa358: the record reads `Cancelled` with `ended_at` set and the
/// machine reports no builder holding a slot.
#[tokio::test]
async fn an_admitted_grant_revives_the_record_its_own_deny_tombstoned_7487() {
    let (state, _dir, session) = hermetic();
    // ADR-0056: version-control holds the main checkout, so the first grant is
    // denied on occupancy.
    insert(
        &state,
        session,
        "version-control",
        "/repo",
        None,
        Some("toolu_vc"),
        DelegationStatus::Running,
    );
    let denied = granted_call(
        &state,
        session,
        dispatch(
            "/repo",
            "rust-engineer",
            Some("worktree"),
            Some("toolu_same"),
        ),
    )
    .await;
    assert!(!denied.claimed, "the occupant denies the first attempt");

    // The occupant finishes and the PM re-dispatches — Claude Code reuses the
    // tool call, so the id is the one the deny just tombstoned.
    let vc = state
        .find_delegation(session, |d| d.tool_use_id.as_deref() == Some("toolu_vc"))
        .expect("the occupant has a record");
    state.terminate_delegation(vc, DelegationStatus::Completed);

    let admitted = granted_call(
        &state,
        session,
        dispatch(
            "/repo",
            "rust-engineer",
            Some("worktree"),
            Some("toolu_same"),
        ),
    )
    .await;
    assert!(admitted.claimed, "the freed checkout admits the grant");

    let record = state
        .delegations_for(session)
        .into_iter()
        .find(|d| d.tool_use_id.as_deref() == Some("toolu_same"))
        .expect("the admitted dispatch has a record");
    assert_eq!(
        record.status,
        DelegationStatus::Running,
        "an admitted dispatch is running, whatever its own deny wrote"
    );
    assert!(
        record.ended_at.is_none(),
        "a revived record carries no end time"
    );
    assert_eq!(record.isolation.as_deref(), Some("worktree"));
}

// ── operator repair of a stuck record (#7602) ────────────────────────────

/// The route is the only way `tm repair delegation` reaches the delegation map,
/// so its path and its plumbing are pinned alongside the gate itself.
#[tokio::test]
async fn repair_route_ends_a_stuck_record_7602() {
    use crate::core::session::{ControlModel, Session, SessionStatus};
    use crate::daemon::services::delegation_repair::{RepairDelegationRequest, RepairOutcome};

    let (state, _dir, session) = hermetic();
    let mut record = Session::new(session, "/repo", ControlModel::Tmux, None);
    record.status = SessionStatus::Stopped;
    state.register_session(record);
    let mut d = Delegation::observed(session, "rust-engineer", "task", Some("toolu_x".into()));
    d.agent_id = Some("af20cc838b2b30a55".to_string());
    d.status = DelegationStatus::Running;
    state.upsert_delegation(d);

    let Json(outcome) = super::repair_delegation_route(
        axum::extract::State(Arc::clone(&state)),
        Path("af20cc838b2b30a55".to_string()),
        Some(Json(RepairDelegationRequest { force: false })),
    )
    .await;

    assert_eq!(outcome, RepairOutcome::Ended { records: 1 });
}

/// A missing body must read as the SAFE default, not as `force: true` — an
/// older client that sends none must not gain the override.
#[tokio::test]
async fn repair_route_refuses_a_live_owner_7602() {
    use crate::core::session::{ControlModel, Session, SessionStatus};
    use crate::daemon::services::delegation_repair::RepairOutcome;

    let (state, _dir, session) = hermetic();
    let mut record = Session::new(session, "/repo", ControlModel::Tmux, None);
    record.status = SessionStatus::Active;
    state.register_session(record);
    let mut d = Delegation::observed(session, "rust-engineer", "task", Some("toolu_x".into()));
    d.agent_id = Some("agent-live".to_string());
    d.status = DelegationStatus::Running;
    state.upsert_delegation(d);

    let Json(outcome) = super::repair_delegation_route(
        axum::extract::State(Arc::clone(&state)),
        Path("agent-live".to_string()),
        None,
    )
    .await;

    assert!(matches!(outcome, RepairOutcome::Refused { .. }));
    assert_eq!(state.all_delegations()[0].status, DelegationStatus::Running);
}

/// A record a stop matched by agent type: live-shaped for a dispatch, no id.
fn type_matched_record(state: &DaemonState, session: SessionId) -> Delegation {
    let mut d = Delegation::observed(session, "version-control", "task", Some("toolu_vc".into()));
    d.cwd = Some(PathBuf::from("/repo"));
    d.status = DelegationStatus::Stale;
    d.stale_by_agent_type = true;
    state.upsert_delegation(d.clone());
    d
}

// #8257: the deny named only `version-control`. The answer now carries the
// record's delegation id (it has no agent id), owner, and clearing command.
#[tokio::test]
async fn shared_tree_dispatch_route_names_each_blocking_record_8257() {
    let (state, _dir, session) = hermetic();
    let d = type_matched_record(&state, session);

    let body = call(
        &state,
        SessionId::new(),
        dispatch("/repo", "rust-engineer", None, Some("toolu_new")),
    )
    .await;

    assert_eq!(body.records.len(), 1, "{body:?}");
    let r = &body.records[0];
    assert_eq!(r.delegation_id, d.id.0.to_string());
    assert_eq!(r.agent_id, None);
    // #8257 owner ruling: the deny JSON the hook reads names no owner UUID.
    assert_eq!(r.owner, "a session the daemon holds no record of");
    crate::daemon::services::delegation_records::delegation_records_tests::assert_no_uuid(
        &serde_json::to_string(&body).expect("json"),
        session,
    );
    assert_eq!(
        r.repair_command,
        format!("tm repair delegation --delegation-id {}", d.id.0)
    );
}

// #8257: the read-only listing names the record the dispatch deny blocks on.
#[tokio::test]
async fn list_route_names_the_blocking_record_8257() {
    let (state, _dir, session) = hermetic();
    let d = type_matched_record(&state, session);

    let Json(listing) = list_delegations_route(
        State(state.clone()),
        Query(ListDelegationsQuery {
            cwd: PathBuf::from("/repo"),
        }),
    )
    .await;

    assert_eq!(listing.records.len(), 1, "{listing:?}");
    assert_eq!(listing.records[0].delegation_id, d.id.0.to_string());
    assert!(listing.records[0].blocks_dispatch);
    // #8257 owner ruling: the listing asks no caller identity, so it names
    // no owner UUID either.
    crate::daemon::services::delegation_records::delegation_records_tests::assert_no_uuid(
        &serde_json::to_string(&listing).expect("json"),
        session,
    );
    assert_eq!(
        state.all_delegations()[0].status,
        DelegationStatus::Stale,
        "listing writes nothing"
    );
}

// #8257: the id-less record is reachable over the wire by its delegation id.
#[tokio::test]
async fn repair_by_id_route_ends_a_record_with_no_agent_id_8257() {
    use crate::core::session::{ControlModel, Session, SessionStatus};
    use crate::daemon::services::delegation_repair::RepairOutcome;

    let (state, _dir, session) = hermetic();
    state.register_session(Session::new(session, "/repo", ControlModel::Tmux, None));
    let d = type_matched_record(&state, session);
    let by_id =
        || repair_delegation_by_id_route(State(state.clone()), Path(d.id.0.to_string()), None);

    // #8257: a type-matched stop may be a sibling's, so an HTTP caller is
    // refused while the owner lives (#8531: HTTP never proves the owner).
    let Json(anonymous) = by_id().await.expect("a well-formed id");
    assert!(
        matches!(anonymous, RepairOutcome::Refused { .. }),
        "{anonymous:?}"
    );
    // Once the owner is gone the record ends by its delegation id.
    let mut stopped = Session::new(session, "/repo", ControlModel::Tmux, None);
    stopped.status = SessionStatus::Stopped;
    state.register_session(stopped);
    let Json(outcome) = by_id().await.expect("a well-formed id");

    assert_eq!(outcome, RepairOutcome::Ended { records: 1 });
    assert_eq!(
        state.all_delegations()[0].status,
        DelegationStatus::Cancelled
    );
    assert!(
        call(
            &state,
            SessionId::new(),
            dispatch("/repo", "rust-engineer", None, Some("t2"))
        )
        .await
        .agents
        .is_empty(),
        "the repaired record no longer blocks a dispatch"
    );
}

// #8257 owner ruling: ownership is never taken from caller-supplied content. A
// request that names the owner's id in the request body — the wire form a CLI
// argument would take — is refused.
#[tokio::test]
async fn repair_route_ignores_an_owner_id_the_caller_supplies_8257() {
    use crate::core::session::{ControlModel, Session};
    use crate::daemon::services::delegation_repair::{RepairDelegationRequest, RepairOutcome};

    let (state, _dir, owner) = hermetic();
    state.register_session(Session::new(owner, "/repo", ControlModel::Tmux, None));
    let mut d = Delegation::observed(owner, "version-control", "task", Some("toolu_c".into()));
    d.agent_id = Some("a0wnclaim".to_string());
    state.upsert_delegation(d);
    let body = || {
        let raw = serde_json::json!({
            "force": false,
            "session": owner.0.to_string(),
            "caller_session": owner.0.to_string(),
            "owner": owner.0.to_string(),
        });
        Some(Json(
            serde_json::from_value::<RepairDelegationRequest>(raw).expect("request body"),
        ))
    };
    let Json(outcome) =
        repair_delegation_route(State(state.clone()), Path("a0wnclaim".to_string()), body()).await;
    assert!(
        matches!(outcome, RepairOutcome::Refused { .. }),
        "{outcome:?}"
    );
    assert!(
        state
            .all_delegations()
            .iter()
            .all(|d| !d.status.is_terminal()),
        "no caller-supplied owner id may clear the record"
    );
}

// #8257 critic R6 Fail-Open Check: the repair runs off the runtime worker, and
// a task that dies before answering is a refusal, never a success.
#[tokio::test]
async fn a_repair_task_that_panics_is_a_refusal_8257() {
    use crate::daemon::services::delegation_repair::RepairOutcome;

    let outcome = repair_off_worker(|| panic!("synthetic repair panic")).await;

    match outcome {
        RepairOutcome::Refused { reason } => {
            assert!(reason.contains("failed before it answered"), "{reason}");
        }
        other => panic!("expected a refusal, got {other:?}"),
    }
}