trusty-mpm 1.8.2

use super::*;
use crate::core::paths::FrameworkPaths;
use crate::core::session::{ControlModel, Session, SessionStatus};
use axum::http::{HeaderMap, StatusCode};
use serial_test::serial;

/// Build a hermetic [`DaemonState`] rooted at an empty temp directory.
///
/// Why: tests that assert overseer-disabled or no-LLM behaviour must not read
/// from the real `~/.trusty-mpm/framework/hooks/overseer.toml`. On a developer
/// machine with `OPENROUTER_API_KEY` set and a live `overseer.toml` the LLM
/// overseer is active, causing two distinct failures (#1523):
///
/// - Tests that assert HTTP 503 ("overseer unavailable") receive HTTP 200 because
///   the real LLM responds.
/// - `ingest_hook` triggers the composite overseer's `pre_tool_use`, which lazily
///   initialises `reqwest::blocking::Client` — that construction internally spawns
///   a new `tokio::Runtime` and panics with "Cannot drop a runtime in a context
///   where blocking is not allowed" when called from a `#[tokio::test]` runtime.
///
/// Both failures vanish when the state reads from a temp dir that contains no
/// `overseer.toml`, so the daemon builds the disabled deterministic overseer.
///
/// What: creates a `tempfile::TempDir`, builds `FrameworkPaths::under` it, calls
/// `DaemonState::with_paths`, and returns both so the caller can keep the
/// `TempDir` alive for the test's duration (the paths are only consulted at
/// construction; in-memory state is self-contained afterwards).
///
/// Test: `hook_relay_runs_with_disabled_overseer`,
/// `non_session_start_event_does_not_auto_register`, `llm_chat_without_overseer_is_503`,
/// `coordinator_chat_without_overseer_is_503`.
fn hermetic_shared() -> (Arc<DaemonState>, tempfile::TempDir) {
    let dir = tempfile::tempdir().expect("temp dir for hermetic DaemonState");
    let paths = FrameworkPaths::under(dir.path());
    let state = Arc::new(DaemonState::with_paths(&paths));
    (state, dir)
}

fn state_with_session() -> (Arc<DaemonState>, SessionId) {
    let (state, _dir) = hermetic_shared();
    let id = SessionId::new();
    let mut session = Session::new(id, "/tmp/p", ControlModel::Tmux, None);
    session.status = SessionStatus::Active;
    state.register_session(session);
    // _dir is intentionally dropped here: FrameworkPaths are only read at
    // construction; the in-memory state is self-contained afterwards.
    (state, id)
}

#[tokio::test]
async fn health_reports_ok_status() {
    // The liveness word stays `ok` — the HR-3 staleness fields are additive.
    let state = DaemonState::shared();
    let Json(body) = health(State(state)).await;
    assert_eq!(body.status, "ok");
}

#[tokio::test]
async fn health_reports_catalog_unknown_without_catalog() {
    // With a fresh framework root and no synced catalog, the staleness check must
    // report `catalog_unknown` (NOT stale, NOT an error) — DOC-17's degrade rule.
    let tmp = tempfile::TempDir::new().unwrap();
    let state = Arc::new(DaemonState::with_root(tmp.path().to_path_buf()));
    let Json(body) = health(State(state)).await;
    assert_eq!(body.status, "ok");
    assert!(
        body.catalog_unknown,
        "no catalog synced → unknown: {body:?}"
    );
    assert!(!body.catalog_stale, "unknown must not be stale");
    assert!(body.catalog_changes.is_empty());
}

#[tokio::test]
async fn health_response_serializes_supervised_field() {
    // #2486: `supervised` must be a real field on the `/health` wire shape,
    // defaulted `true` (safe) until `daemon_run` computes and sets it.
    let state = DaemonState::shared();
    let Json(body) = health(State(state)).await;
    assert!(
        body.supervised,
        "supervised defaults true before daemon_run sets it"
    );

    let value = serde_json::to_value(&body).expect("HealthResponse must serialize");
    assert_eq!(
        value.get("supervised"),
        Some(&serde_json::Value::Bool(true)),
        "wire shape must carry `supervised`: {value}"
    );
}

/// #8058: `/health` must carry the `degraded` list, and it must be EMPTY on a
/// daemon whose pr-cleanup sweep has not parked itself.
///
/// Why: the field exists so a suspended subsystem is visible. A field that is
/// non-empty by accident would train readers to ignore it, which is the same
/// failure as not having it.
/// Test: this test.
#[tokio::test]
async fn health_publishes_no_degraded_subsystem_when_idle() {
    let state = DaemonState::shared();
    let Json(body) = health(State(state)).await;
    assert!(
        body.degraded.is_empty(),
        "an idle daemon reports nothing degraded: {:?}",
        body.degraded
    );

    let value = serde_json::to_value(&body).expect("HealthResponse must serialize");
    assert_eq!(
        value.get("degraded"),
        Some(&serde_json::Value::Array(Vec::new())),
        "wire shape must carry `degraded`: {value}"
    );
}

/// #4469: `/health` must publish the THREE-STATE launchd answer, not only the
/// bool that collapses it.
///
/// Why: without this field a launchctl that could not be asked is indistinguishable
/// on the wire from launchd positively saying "not mine", and `tm doctor`'s orphan
/// verdict escalates the latter to a `kill -TERM` recommendation. The fix has to
/// REACH the surface that motivated it.
/// Test: this test.
#[tokio::test]
async fn health_response_serializes_launchd_supervision_field() {
    let state = DaemonState::shared();
    state.set_launchd_supervision("unknown");
    let Json(body) = health(State(state)).await;
    assert_eq!(body.launchd_supervision, "unknown");

    let value = serde_json::to_value(&body).expect("HealthResponse must serialize");
    assert_eq!(
        value.get("launchd_supervision"),
        Some(&serde_json::Value::String("unknown".into())),
        "wire shape must carry `launchd_supervision`: {value}"
    );
}

#[tokio::test]
async fn health_response_serializes_pid_field() {
    // #4230: `supervised` is a self-report (a `getppid() == 1` heuristic before #4469),
    // so an orphan can self-report `true`. `pid` is what makes `tm doctor`'s check
    // authoritative — it identifies WHICH process answered, for comparison against
    // the PID launchd reports for the registered daemon label.
    let state = DaemonState::shared();
    let Json(body) = health(State(state)).await;
    assert_eq!(
        body.pid,
        std::process::id(),
        "pid must identify the responding process"
    );

    let value = serde_json::to_value(&body).expect("HealthResponse must serialize");
    assert_eq!(
        value.get("pid").and_then(serde_json::Value::as_u64),
        Some(u64::from(std::process::id())),
        "wire shape must carry `pid`: {value}"
    );
}

#[tokio::test]
async fn health_response_serializes_forced_field() {
    // #4230: distinguishes a deliberate `tm daemon --force` run from an unwanted
    // orphan, so `tm doctor` does not hard-Fail on the escape hatch its own
    // remediation recommends. Defaults `false` until `daemon_run` sets it.
    let state = DaemonState::shared();
    let Json(body) = health(State(state)).await;
    assert!(
        !body.unsupervised_forced,
        "force flag defaults false before daemon_run sets it"
    );

    let value = serde_json::to_value(&body).expect("HealthResponse must serialize");
    assert_eq!(
        value.get("unsupervised_forced"),
        Some(&serde_json::Value::Bool(false)),
        "wire shape must carry `unsupervised_forced`: {value}"
    );
}

/// #7822: `/health` must carry a BUILD fingerprint, not only a version — two
/// builds cut under one semver are indistinguishable through `version`, and the
/// daemon that outlives a merge is exactly that case. Before `daemon_run`
/// records it the value is empty, which the client reads as "cannot tell".
#[tokio::test]
async fn health_response_serializes_build_id_field() {
    let state = DaemonState::shared();
    assert_eq!(
        state.build_identity(),
        "",
        "empty until daemon_run records the startup fingerprint"
    );
    state.set_build_identity("1757731440:74125312");
    let Json(body) = health(State(state)).await;
    assert_eq!(body.build_id, "1757731440:74125312");

    let value = serde_json::to_value(&body).expect("HealthResponse must serialize");
    assert_eq!(
        value.get("build_id"),
        Some(&serde_json::Value::String("1757731440:74125312".to_owned())),
        "wire shape must carry `build_id`: {value}"
    );
}

#[tokio::test]
async fn health_response_serializes_version_field() {
    // Issue #2332: `/health` must carry this process's build version so a
    // client can detect a stale daemon that predates the installed binary.
    let state = DaemonState::shared();
    let Json(body) = health(State(state)).await;
    assert_eq!(body.version, env!("CARGO_PKG_VERSION"));

    let value = serde_json::to_value(&body).expect("HealthResponse must serialize");
    assert_eq!(
        value.get("version"),
        Some(&serde_json::Value::String(
            env!("CARGO_PKG_VERSION").to_owned()
        )),
        "wire shape must carry `version`: {value}"
    );
}

#[tokio::test]
async fn current_project_found_and_missing() {
    // `GET /projects/current` returns the project for a registered path
    // and `404` for an unregistered one.
    let state = DaemonState::shared();
    let _ = register_project(
        State(Arc::clone(&state)),
        Json(RegisterProject {
            path: "/work/demo".into(),
        }),
    )
    .await;

    let ok = current_project(
        State(Arc::clone(&state)),
        Query(CurrentProjectQuery {
            path: "/work/demo".into(),
        }),
    )
    .await;
    assert!(ok.is_ok());

    let err = current_project(
        State(state),
        Query(CurrentProjectQuery {
            path: "/work/missing".into(),
        }),
    )
    .await
    .unwrap_err();
    assert_eq!(err.status(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn discover_projects_returns_array() {
    // `GET /projects/discover` always answers with a (possibly empty) array;
    // it must never error even when `~/.claude/projects/` is absent.
    let state = DaemonState::shared();
    let resp = discover_projects(State(state)).await;
    // The discovered list is well-formed; on CI it is typically empty.
    for project in &resp.0.projects {
        assert!(!project.path.is_empty());
    }
}

#[tokio::test]
async fn register_session_associates_project() {
    // A `POST /sessions` body carrying `project_path` must associate the
    // new session with that project.
    let state = DaemonState::shared();
    let Json(body) = register_session(
        State(Arc::clone(&state)),
        Json(RegisterSession {
            project: "/work/demo".into(),
            project_path: Some("/work/demo".into()),
            name: None,
            workdir: None,
        }),
    )
    .await
    .expect("registration-only path succeeds");
    let id = body.id.0.to_string();
    let listed = state.list_sessions();
    let session = listed
        .iter()
        .find(|s| s.id.0.to_string() == id)
        .expect("session registered");
    assert_eq!(session.project_path, Some(PathBuf::from("/work/demo")));
}

#[tokio::test]
async fn list_sessions_filters_by_project() {
    // `GET /sessions?project=<path>` returns only sessions of that project.
    let state = DaemonState::shared();
    let _ = register_session(
        State(Arc::clone(&state)),
        Json(RegisterSession {
            project: "/work/demo".into(),
            project_path: Some("/work/demo".into()),
            name: None,
            workdir: None,
        }),
    )
    .await;
    let _ = register_session(
        State(Arc::clone(&state)),
        Json(RegisterSession {
            project: "/work/other".into(),
            project_path: Some("/work/other".into()),
            name: None,
            workdir: None,
        }),
    )
    .await;

    let Json(all) = list_sessions(State(Arc::clone(&state)), Query(SessionQuery::default())).await;
    assert_eq!(all.sessions.len(), 2);

    let Json(scoped) = list_sessions(
        State(state),
        Query(SessionQuery {
            project: Some("/work/demo".into()),
        }),
    )
    .await;
    assert_eq!(scoped.sessions.len(), 1);
}

#[tokio::test]
async fn hook_relay_ingests_known_event() {
    let (state, id) = state_with_session();
    let post = HookPost {
        session_id: id.0.to_string(),
        event: HookEvent::PostToolUse,
        payload: serde_json::json!({"tool": "Edit"}),
    };
    let result = ingest_hook(State(state.clone()), Json(post)).await;
    assert!(result.is_ok());
    assert_eq!(state.recent_hook_events().len(), 1);
}

#[tokio::test]
async fn register_and_remove_session() {
    let state = DaemonState::shared();
    let Json(body) = register_session(
        State(state.clone()),
        Json(RegisterSession {
            project: "/tmp/new".into(),
            project_path: None,
            name: None,
            workdir: None,
        }),
    )
    .await
    .expect("registration-only path succeeds");
    let id = body.id.0.to_string();
    assert_eq!(state.list_sessions().len(), 1);
    // Removing it succeeds; removing again is a 404.
    assert!(
        remove_session(State(state.clone()), Path(id.clone()))
            .await
            .is_ok()
    );
    let err = remove_session(State(state), Path(id)).await.unwrap_err();
    assert_eq!(err.status(), StatusCode::NOT_FOUND);
}

/// Register one legacy-registry record under `tmux_name`, in the shape #7834
/// reproduced: `status: Starting`, no managed `state`/`slot`/`source_id`.
///
/// Why: an in-place `tm session start --dir <dir>` against a clone whose remote
/// is not GitHub lands here, not in the managed store — that is the record
/// stop-by-name has to resolve.
/// Test: `remove_session_by_name_removes_legacy_registry_record`,
/// `remove_session_by_id_still_removes_legacy_registry_record`.
fn register_legacy_record(state: &DaemonState, tmux_name: &str) -> SessionId {
    let id = SessionId::new();
    let mut session = Session::new(id, "/tmp/p", ControlModel::Tmux, None);
    session.status = SessionStatus::Starting;
    session.tmux_name = tmux_name.to_string();
    state.register_session(session);
    id
}

#[tokio::test]
async fn remove_session_by_name_removes_legacy_registry_record() {
    // #7834: `session info <name>` resolved the friendly name while DELETE
    // parsed the path segment strictly as a UUID and returned 400. Stop-by-name
    // must resolve the legacy record AND run the teardown against the RESOLVED
    // record's `tmux_name` — asserted here through the managed-store reconcile,
    // the one teardown step observable without a live tmux server.
    let ws = std::path::PathBuf::from("/tmp/test-ws-7834-stop-by-name");
    let (state, managed_id, tmux_name) = make_state_with_active_managed(ws).await;
    let id = register_legacy_record(&state, &tmux_name);

    let Json(body) = remove_session(State(Arc::clone(&state)), Path(tmux_name.clone()))
        .await
        .expect("stop by friendly name must resolve a legacy-registry record");

    assert_eq!(
        body.removed, tmux_name,
        "the response echoes what was asked"
    );
    assert!(
        state.session(id).is_none(),
        "the legacy registry entry must be gone after stop-by-name"
    );
    let mgr = state.session_manager().await;
    let record = mgr.get(&managed_id).await.expect("managed record readable");
    assert_eq!(
        record.state,
        crate::session_manager::ManagedSessionState::Decommissioned,
        "teardown must run against the resolved record's tmux_name"
    );
}

#[tokio::test]
async fn remove_session_by_id_still_removes_legacy_registry_record() {
    // The #7834 workaround (stop by uuid) keeps working — name resolution is
    // additive, not a replacement for the UUID form.
    let (state, _dir) = hermetic_shared();
    let id = register_legacy_record(&state, "tm-verify-clone");

    let Json(body) = remove_session(State(Arc::clone(&state)), Path(id.0.to_string()))
        .await
        .expect("stop by uuid must still remove a legacy-registry record");

    assert_eq!(body.removed, id.0.to_string());
    assert!(state.session(id).is_none(), "the entry must be gone");
}

#[tokio::test]
async fn remove_session_unknown_name_is_404() {
    // A name that matches nothing is a 404 naming the session, never the 400
    // #7834 reported — "not found" and "malformed" are different answers.
    let (state, _dir) = hermetic_shared();
    let err = remove_session(State(state), Path("tm-no-such-session".to_string()))
        .await
        .unwrap_err();

    assert_eq!(err.status(), StatusCode::NOT_FOUND);
    assert!(
        err.to_string().contains("tm-no-such-session"),
        "the refusal must name the session: {err}"
    );
}

#[tokio::test]
async fn get_session_returns_session() {
    // `GET /sessions/{id}` resolves a single session by id and returns its
    // current snapshot so callers don't have to page through `GET /sessions`.
    let (state, id) = state_with_session();
    let Json(session) = get_session(State(state), Path(id.0.to_string()))
        .await
        .expect("known id resolves");
    assert_eq!(session.id, id);
    assert_eq!(session.workdir, "/tmp/p");
}

#[tokio::test]
async fn get_session_unknown_is_404() {
    // An unknown UUID is a 404, matching the rest of the sessions surface.
    let state = DaemonState::shared();
    let err = get_session(State(state), Path(SessionId::new().0.to_string()))
        .await
        .unwrap_err();
    assert_eq!(err.status(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn get_session_malformed_id_is_400() {
    // A non-UUID id is a 400 before the lookup runs, mirroring `parse_id`.
    let state = DaemonState::shared();
    let err = get_session(State(state), Path("not-a-uuid".to_string()))
        .await
        .unwrap_err();
    assert_eq!(err.status(), StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn connect_session_registers_without_deploy() {
    // `POST /api/v1/sessions/connect` performs the same daemon-side
    // bookkeeping as `POST /sessions` — it registers the session and returns
    // its id and friendly name. The daemon never deploys framework artifacts
    // in either path; the connect/launch distinction lives in the client.
    let state = DaemonState::shared();
    let Json(body) = connect_session(
        State(Arc::clone(&state)),
        Json(RegisterSession {
            project: "/tmp/connect".into(),
            project_path: Some("/tmp/connect".into()),
            name: Some("tmpm-connect".into()),
            workdir: None,
        }),
    )
    .await
    .expect("connect registration succeeds");
    assert_eq!(body.name, "tmpm-connect");
    let listed = state.list_sessions();
    let session = listed
        .iter()
        .find(|s| s.id == body.id)
        .expect("session registered via connect");
    assert_eq!(session.workdir, "/tmp/connect");
}

#[tokio::test]
async fn registered_session_has_friendly_tmux_name() {
    // A registered session must carry a `tmpm-<adj>-<noun>` tmux name
    // derived from its UUID, not the legacy `trusty-mpm-<uuid>` form.
    let state = DaemonState::shared();
    let Json(body) = register_session(
        State(Arc::clone(&state)),
        Json(RegisterSession {
            project: "/tmp/friendly".into(),
            project_path: None,
            name: None,
            workdir: None,
        }),
    )
    .await
    .expect("registration-only path succeeds");
    let id = body.id.0.to_string();
    let listed = state.list_sessions();
    let session = listed
        .iter()
        .find(|s| s.id.0.to_string() == id)
        .expect("session registered");
    assert!(
        session.tmux_name.starts_with("tm-"),
        "friendly name: {}",
        session.tmux_name
    );
    assert!(session.tmux_name.len() <= 25);
}

#[tokio::test]
async fn reap_sessions_returns_removed_count() {
    // `DELETE /sessions/dead` always returns a well-formed `{ "removed": N }`
    // body. The exact count depends on whether tmux is installed: with tmux
    // the lone test session (no live tmux session named `tmpm-*`) is reaped
    // (1); without tmux nothing is reaped (0). Either way the registry must
    // not contain a session that is missing from tmux afterwards.
    let (state, _) = state_with_session();
    let Json(body) = reap_sessions(State(Arc::clone(&state))).await;
    let removed = body.removed;
    assert!(removed <= 1, "at most the one test session is reaped");
    assert_eq!(state.list_sessions().len(), 1 - removed);
}

#[tokio::test]
async fn spawn_session_without_claude_returns_422() {
    // `POST /sessions` with a `workdir` opts into spawn mode. When the
    // `claude` binary is unavailable, the handler must return HTTP 422 (and
    // leave the session registry empty — no half-created bookkeeping).
    let _claude = crate::daemon::services::tmux_service::set_claude_lookup_override(Some(None));
    let state = DaemonState::shared();
    let err = register_session(
        State(Arc::clone(&state)),
        Json(RegisterSession {
            project: "/tmp/spawn-no-claude".into(),
            project_path: Some("/tmp/spawn-no-claude".into()),
            name: None,
            workdir: Some("/tmp/spawn-no-claude".into()),
        }),
    )
    .await
    .expect_err("spawn mode without claude must error");
    assert_eq!(err.status(), StatusCode::UNPROCESSABLE_ENTITY);
    assert!(
        state.list_sessions().is_empty(),
        "no session should be registered on 422"
    );
}

/// Why serial (#5813): every tmux path in this crate runs through
/// `TmuxDriver::discover`, which consults the #5784 host-state gate. That gate
/// reads the PROCESS `$HOME` and compares it against the password-database
/// home for this uid — so while any of the ~40 `$HOME`-overriding tests in this
/// binary holds its temp home, `discover` classifies a scratch environment and
/// returns `DaemonError::TmuxUnavailable`, which is a 500. The assertion below
/// then reads `left: 500, right: 422` for a reason that has nothing to do with
/// tmux. `#[serial]`'s crate-wide default group is the remedy this crate
/// already documents for that global — see `test_support::tests::HomeOverride`.
#[tokio::test]
#[serial]
async fn spawn_session_without_tmux_returns_422_on_no_tmux_host() {
    // Force the `claude` lookup positive so the spawn proceeds past the
    // binary check, then assert that the daemon degrades gracefully when
    // tmux is unavailable. On CI tmux is generally absent, in which case the
    // documented 422 applies. On a developer host that *does* have tmux
    // installed the spawn will either succeed or surface an internal tmux
    // error — both are acceptable shapes; the contract this test enforces
    // is "never panic, and 422 when tmux missing".
    let _claude = crate::daemon::services::tmux_service::set_claude_lookup_override(Some(Some(
        "/fake/claude".into(),
    )));
    let state = DaemonState::shared();
    let outcome = register_session(
        State(Arc::clone(&state)),
        Json(RegisterSession {
            project: "/tmp/spawn-no-tmux".into(),
            project_path: Some("/tmp/spawn-no-tmux".into()),
            name: Some("tmpm-spawn-test-no-tmux".into()),
            workdir: Some("/tmp".into()),
        }),
    )
    .await;
    if crate::daemon::tmux::TmuxDriver::is_available() {
        // On a tmux-equipped host the spawn either succeeds or errors with an
        // internal error from the bogus `claude` path; clean up if it created
        // a real session, then return.
        if let Ok(driver) = crate::daemon::tmux::TmuxDriver::discover() {
            let _ = driver.kill_session("tmpm-spawn-test-no-tmux");
        }
        // Either way the registry must not contain a session for a failed
        // spawn — successful spawns leave one entry; remove it for hygiene.
        for s in state.list_sessions() {
            state.remove_session(s.id);
        }
        let _ = outcome;
    } else {
        let err = outcome.expect_err("spawn mode without tmux must error");
        assert_eq!(err.status(), StatusCode::UNPROCESSABLE_ENTITY);
        assert!(
            state.list_sessions().is_empty(),
            "no session should be registered on 422"
        );
    }
}

#[tokio::test]
async fn registration_only_path_ignores_missing_claude() {
    // The bookkeeping (registration-only) path must NOT consult `claude` —
    // an absent binary is irrelevant when no spawn was requested. Forcing
    // the lookup negative proves the field is the sole trigger.
    let _claude = crate::daemon::services::tmux_service::set_claude_lookup_override(Some(None));
    let state = DaemonState::shared();
    let Json(body) = register_session(
        State(Arc::clone(&state)),
        Json(RegisterSession {
            project: "/tmp/no-spawn".into(),
            project_path: None,
            name: None,
            workdir: None,
        }),
    )
    .await
    .expect("registration-only path must succeed regardless of claude availability");
    assert_eq!(state.list_sessions().len(), 1);
    assert_eq!(state.list_sessions()[0].id, body.id);
}

#[tokio::test]
async fn register_session_returns_id_even_without_tmux() {
    // Graceful-degradation invariant: tmux is unavailable in CI, yet
    // `POST /sessions` must still return a JSON body carrying an `id`, and
    // that id must be visible in the subsequent `GET /sessions` snapshot.
    let state = DaemonState::shared();
    let Json(body) = register_session(
        State(Arc::clone(&state)),
        Json(RegisterSession {
            project: "/tmp/no-tmux".into(),
            project_path: None,
            name: None,
            workdir: None,
        }),
    )
    .await
    .expect("registration-only path succeeds");
    let id_str = body.id.0.to_string();
    let listed = state.list_sessions();
    assert_eq!(listed.len(), 1);
    assert_eq!(listed[0].id.0.to_string(), id_str);
}

#[tokio::test]
async fn hook_relay_rejects_bad_session_id() {
    let (state, _) = state_with_session();
    let post = HookPost {
        session_id: "not-a-uuid".into(),
        event: HookEvent::Stop,
        payload: serde_json::Value::Null,
    };
    let err = ingest_hook(State(state), Json(post)).await.unwrap_err();
    assert_eq!(err.status(), StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn hook_relay_runs_with_disabled_overseer() {
    // With the overseer disabled (the default), a PreToolUse event must
    // still be ingested normally — the overseer fast-path allows it.
    let (state, id) = state_with_session();
    let post = HookPost {
        session_id: id.0.to_string(),
        event: HookEvent::PreToolUse,
        payload: serde_json::json!({"tool": "Bash", "input": {"command": "ls"}}),
    };
    let result = ingest_hook(State(state.clone()), Json(post)).await;
    assert!(result.is_ok());
    assert_eq!(state.recent_hook_events().len(), 1);
}

#[tokio::test]
async fn session_start_auto_registers_unknown_session() {
    // A `SessionStart` hook for a session the daemon has never seen must
    // auto-register it (connection-driven registration), using the incoming
    // UUID so the session carries the right id.
    let state = DaemonState::shared();
    let new_id = crate::core::session::SessionId::new();
    assert!(state.session(new_id).is_none());

    let post = HookPost {
        session_id: new_id.0.to_string(),
        event: HookEvent::SessionStart,
        payload: serde_json::Value::Null,
    };
    let result = ingest_hook(State(state.clone()), Json(post)).await;
    assert!(result.is_ok());

    let registered = state.session(new_id).expect("session auto-registered");
    assert_eq!(registered.id, new_id);
    assert_eq!(
        registered.status,
        crate::core::session::SessionStatus::Active
    );
}

#[tokio::test]
async fn non_session_start_event_does_not_auto_register() {
    // Only `SessionStart` auto-registers. A non-start event for an unknown
    // session must not create a session record. Uses hermetic state: a
    // `PreToolUse` event triggers the LLM overseer when one is configured via
    // `~/.trusty-mpm/overseer.toml`; the blocking reqwest client init then
    // panics inside the tokio test runtime (#1523).
    let (state, _dir) = hermetic_shared();
    let unknown = crate::core::session::SessionId::new();
    let post = HookPost {
        session_id: unknown.0.to_string(),
        event: HookEvent::PreToolUse,
        payload: serde_json::json!({"tool": "Bash"}),
    };
    let _ = ingest_hook(State(state.clone()), Json(post)).await;
    assert!(state.session(unknown).is_none());
}

#[tokio::test]
async fn session_start_for_known_session_does_not_duplicate() {
    // A `SessionStart` for an already-registered session must not create a
    // second record.
    let (state, id) = state_with_session();
    let before = state.list_sessions().len();
    let post = HookPost {
        session_id: id.0.to_string(),
        event: HookEvent::SessionStart,
        payload: serde_json::Value::Null,
    };
    let result = ingest_hook(State(state.clone()), Json(post)).await;
    assert!(result.is_ok());
    assert_eq!(state.list_sessions().len(), before);
}

#[tokio::test]
async fn llm_chat_without_overseer_is_503() {
    // A daemon built without an overseer config (hermetic empty temp dir) must
    // report the LLM chat capability as unavailable — the endpoint can only
    // route to the LLM when an API key resolved at startup (#1523).
    let (state, _dir) = hermetic_shared();
    let err = llm_chat(
        State(state),
        Json(LlmChatRequest {
            message: "hello".into(),
            history: Vec::new(),
        }),
    )
    .await
    .unwrap_err();
    assert_eq!(err.status(), StatusCode::SERVICE_UNAVAILABLE);
}

#[tokio::test]
async fn coordinator_context_returns_snapshot() {
    // `GET /api/v1/sessions/context` always returns a snapshot; with a
    // registered session it appears in the `sessions` array.
    let (state, _id) = state_with_session();
    let snapshot = coordinator_context(State(state)).await;
    assert_eq!(snapshot.sessions.len(), 1);
}

#[tokio::test]
async fn coordinator_chat_without_overseer_is_503() {
    // A daemon built without an overseer config (hermetic empty temp dir) has
    // neither LLM overseer nor Session Manager runtime, so a non-prefixed
    // coordinator message must return HTTP 503 (#1523).
    let (state, _dir) = hermetic_shared();
    let err = coordinator_chat(
        State(state),
        HeaderMap::new(),
        Json(CoordinatorChatRequest {
            message: "what is happening?".into(),
            history: Vec::new(),
            conv_id: None,
            actions: None,
        }),
    )
    .await
    .unwrap_err();
    assert_eq!(err.status(), StatusCode::SERVICE_UNAVAILABLE);
}

#[tokio::test]
async fn coordinator_chat_routes_prefixed_message() {
    // A `@prefix:` message routes directly to the session's tmux pane and
    // never touches the LLM, so it succeeds even with no API key configured.
    let state = DaemonState::shared();
    let id = SessionId::new();
    let mut session = Session::new(id, "/tmp/p", ControlModel::Tmux, None);
    session.status = SessionStatus::Active;
    session.tmux_name = "tmpm-coordtest".to_string();
    state.register_session(session);

    let resp = coordinator_chat(
        State(state),
        HeaderMap::new(),
        Json(CoordinatorChatRequest {
            message: "@coordtest: echo hi".into(),
            history: Vec::new(),
            conv_id: None,
            actions: None,
        }),
    )
    .await
    .expect("prefixed routing must not require an LLM");
    assert_eq!(resp.routed_to_session.as_deref(), Some("tmpm-coordtest"));
    assert!(resp.command_output.is_some());
}

// The /api-docs routes only exist when the `swagger-ui` feature is active.
#[cfg(feature = "swagger-ui")]
#[tokio::test]
async fn openapi_spec_is_valid() {
    // `GET /api-docs/openapi.json` must return 200 with a document that
    // carries the `openapi` version key and the daemon's title.
    use axum::body::Body;
    use axum::http::Request;
    use tower::ServiceExt;

    let app = router(DaemonState::shared());
    let response = app
        .oneshot(
            Request::builder()
                .uri("/api-docs/openapi.json")
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::OK);

    let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
        .await
        .unwrap();
    let spec: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
    assert!(
        spec.get("openapi").is_some(),
        "spec must have an openapi key"
    );
    assert!(
        spec["info"]["title"]
            .as_str()
            .unwrap_or_default()
            .contains("trusty-mpm"),
        "spec title must mention trusty-mpm"
    );
}

#[tokio::test]
async fn rpc_rejects_non_loopback_peer() {
    // #1221 hard requirement: POST /rpc must 403 a non-loopback source IP even
    // though the daemon binds loopback — guarding a future 0.0.0.0 rebind.
    use axum::body::Body;
    use axum::extract::ConnectInfo;
    use axum::http::Request;
    use std::net::{IpAddr, Ipv4Addr, SocketAddr};
    use tower::ServiceExt;

    let app = router(DaemonState::shared());
    // Simulate a request arriving from a public LAN address.
    let peer = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 50)), 40000);
    let mut req = Request::builder()
        .method("POST")
        .uri("/rpc")
        .header("content-type", "application/json")
        .body(Body::from(
            r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#,
        ))
        .unwrap();
    req.extensions_mut().insert(ConnectInfo(peer));

    let response = app.oneshot(req).await.unwrap();
    assert_eq!(
        response.status(),
        StatusCode::FORBIDDEN,
        "non-loopback /rpc must be forbidden"
    );
}

#[tokio::test]
async fn rpc_dispatches_tools_list_for_loopback() {
    // A loopback peer is allowed and gets the full tool catalog back through the
    // in-process MCP dispatch path. We assert the catalog CONTAINS the expected
    // tool NAMES (not a brittle exact count): the 6 new session-lifecycle tools
    // must all be present, and the total must be at least 15. The expected
    // breakdown is 9 pre-existing tools + 6 new session-lifecycle tools = 15; if
    // a future change adds tools, only this comment and the `>= 15` floor need
    // revisiting — the name assertions keep protecting the session surface.
    use axum::body::Body;
    use axum::extract::ConnectInfo;
    use axum::http::Request;
    use std::net::{IpAddr, Ipv4Addr, SocketAddr};
    use tower::ServiceExt;

    let app = router(DaemonState::shared());
    let peer = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 50505);
    let mut req = Request::builder()
        .method("POST")
        .uri("/rpc")
        .header("content-type", "application/json")
        .body(Body::from(
            r#"{"jsonrpc":"2.0","id":7,"method":"tools/list"}"#,
        ))
        .unwrap();
    req.extensions_mut().insert(ConnectInfo(peer));

    let response = app.oneshot(req).await.unwrap();
    assert_eq!(response.status(), StatusCode::OK);

    let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
        .await
        .unwrap();
    let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
    assert_eq!(body["id"], 7);
    let tools = body["result"]["tools"].as_array().expect("tools array");

    // Collect the advertised tool names for name-based (not count-based) asserts.
    let names: Vec<&str> = tools.iter().filter_map(|t| t["name"].as_str()).collect();

    // The 6 new session-lifecycle tools must all be advertised.
    for expected in [
        "session_new",
        "session_stop",
        "session_resume",
        "session_decommission",
        "session_activity",
        "session_send",
    ] {
        assert!(
            names.contains(&expected),
            "loopback /rpc tools/list must advertise `{expected}`; got {names:?}"
        );
    }

    // Count floor: 9 existing + 6 new = 15. Use `>=` so adding future tools does
    // not break this test for the wrong reason.
    assert!(
        tools.len() >= 15,
        "loopback /rpc tools/list must advertise at least 15 tools (9 existing + 6 session-lifecycle); got {}",
        tools.len()
    );
}

#[tokio::test]
async fn pause_then_resume_round_trips() {
    // Pausing flips a session to `Paused`; resuming flips it back to
    // `Active` and clears the pause metadata.
    let (state, id) = state_with_session();
    let Json(body) = pause_session(
        State(Arc::clone(&state)),
        Path(id.0.to_string()),
        Json(PauseRequest {
            summary: Some("mid-task".into()),
        }),
    )
    .await
    .expect("pause succeeds");
    assert!(body.paused);
    assert_eq!(body.summary, "mid-task");

    let paused = state.session(id).expect("session exists");
    assert_eq!(paused.status, SessionStatus::Paused);
    assert_eq!(paused.pause_summary.as_deref(), Some("mid-task"));
    assert!(paused.paused_at.is_some());

    let Json(resumed) = resume_session(State(Arc::clone(&state)), Path(id.0.to_string()))
        .await
        .expect("resume succeeds");
    assert!(resumed.resumed);

    let active = state.session(id).expect("session exists");
    assert_eq!(active.status, SessionStatus::Active);
    assert_eq!(active.paused_at, None);
    assert_eq!(active.pause_summary, None);
}

#[tokio::test]
async fn pause_unknown_session_is_404() {
    let state = DaemonState::shared();
    let err = pause_session(
        State(state),
        Path(SessionId::new().0.to_string()),
        Json(PauseRequest::default()),
    )
    .await
    .unwrap_err();
    assert_eq!(err.status(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn resume_unpaused_session_is_409() {
    // A session that was never paused cannot be resumed.
    let (state, id) = state_with_session();
    let err = resume_session(State(state), Path(id.0.to_string()))
        .await
        .unwrap_err();
    assert_eq!(err.status(), StatusCode::CONFLICT);
}

#[tokio::test]
async fn command_to_stopped_session_is_409() {
    let state = DaemonState::shared();
    let id = SessionId::new();
    let mut session = Session::new(id, "/tmp/p", ControlModel::Tmux, None);
    session.status = SessionStatus::Stopped;
    state.register_session(session);

    let err = send_command(
        State(state),
        Path(id.0.to_string()),
        Query(CommandQuery::default()),
        Json(CommandRequest {
            command: "help".into(),
        }),
    )
    .await
    .unwrap_err();
    assert_eq!(err.status(), StatusCode::CONFLICT);
}

#[tokio::test]
async fn output_unknown_session_is_404() {
    let state = DaemonState::shared();
    let err = get_output(
        State(state),
        Path(SessionId::new().0.to_string()),
        Query(OutputQuery::default()),
    )
    .await
    .unwrap_err();
    assert_eq!(err.status(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn pause_resolves_session_by_friendly_name() {
    // The pause endpoint accepts a friendly tmux name, not just a UUID.
    let (state, id) = state_with_session();
    let name = state.session(id).expect("session").tmux_name;
    let Json(body) = pause_session(
        State(Arc::clone(&state)),
        Path(name),
        Json(PauseRequest::default()),
    )
    .await
    .expect("pause by name succeeds");
    assert!(body.paused);
}

#[test]
fn send_command_compress_query_defaults_off() {
    // A `CommandQuery` with no `compress` field deserializes to `None`, so
    // omitting `?compress=` defaults to no compression.
    let query: CommandQuery = serde_json::from_str("{}").expect("empty query deserializes");
    assert_eq!(query.compress, None);
}

#[test]
fn output_query_defaults() {
    // An `OutputQuery` with no fields set has neither a line count nor a
    // compression level.
    let query: OutputQuery = serde_json::from_str("{}").expect("empty query deserializes");
    assert_eq!(query.lines, None);
    assert_eq!(query.compress, None);
}

#[test]
fn compress_level_roundtrips_serde() {
    // `CompressionLevel::Summarise` serializes to the lowercase wire name
    // `"summarise"` and deserializes back to the same variant.
    let json = serde_json::to_string(&CompressionLevel::Summarise).expect("serialize");
    assert_eq!(json, "\"summarise\"");
    let parsed: CompressionLevel = serde_json::from_str(&json).expect("deserialize");
    assert_eq!(parsed, CompressionLevel::Summarise);
}

#[test]
fn compress_level_label_matches_serde() {
    // The lowercase label helper agrees with serde's wire representation.
    assert_eq!(compression_level_label(CompressionLevel::Off), "off");
    assert_eq!(compression_level_label(CompressionLevel::Trim), "trim");
    assert_eq!(
        compression_level_label(CompressionLevel::Summarise),
        "summarise"
    );
    assert_eq!(
        compression_level_label(CompressionLevel::Caveman),
        "caveman"
    );
}

#[test]
fn apply_compression_off_is_passthrough() {
    // With no level, the text is returned unchanged and there is no label.
    let result = apply_compression(None, "raw pane text");
    assert_eq!(result.text, "raw pane text");
    assert_eq!(result.level_label, None);
}

#[test]
fn apply_compression_summarise() {
    // With a level set, the label is recorded and stats reflect the input.
    let raw = "x".repeat(100);
    let result = apply_compression(Some(CompressionLevel::Summarise), &raw);
    assert_eq!(result.level_label.as_deref(), Some("summarise"));
    assert_eq!(result.stats.original_bytes, 100);
}

/// Why serial (#5813): `TmuxService::adopt` shares
/// `tmux_snapshot_unknown_session_is_404`'s exposure to the host-state gate —
/// it also reaches tmux through `discover_or_session_error`, so a concurrent
/// `$HOME` override turns this 404 into a 500.
#[tokio::test]
#[serial]
async fn adopt_tmux_session_handles_missing() {
    // Adopting a session that does not exist (or with tmux absent) is 404.
    let state = DaemonState::shared();
    let result = adopt_tmux_session(
        State(state),
        Json(AdoptRequest {
            session: "trusty-mpm-no-such-session-xyz".into(),
        }),
    )
    .await;
    assert_eq!(result.unwrap_err().status(), StatusCode::NOT_FOUND);
}

/// Why serial (#5813): same `$HOME`-vs-host-state-gate race as
/// `spawn_session_without_tmux_returns_422_on_no_tmux_host` — a concurrent
/// `$HOME` override makes `TmuxService::snapshot` return `TmuxUnavailable`
/// (500) instead of the `SessionNotFound` (404) this asserts.
#[tokio::test]
#[serial]
async fn tmux_snapshot_unknown_session_is_404() {
    let state = DaemonState::shared();
    let result = tmux_snapshot(State(state), Path("no-such-session-xyz".into())).await;
    assert_eq!(result.unwrap_err().status(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn create_checkpoint_returns_id() {
    // `POST /claude-config/checkpoints` returns an `id` and the checkpoint
    // is then visible via the list endpoint.
    let dir = tempfile::tempdir().unwrap();
    let state = DaemonState::shared();
    let Json(body) = create_checkpoint(
        State(Arc::clone(&state)),
        Json(CreateCheckpointRequest {
            project: dir.path().to_path_buf(),
            label: Some("manual".into()),
        }),
    )
    .await
    .expect("create succeeds");
    assert!(!body.id.is_empty());

    let Json(listed) = list_checkpoints(
        State(state),
        Query(CheckpointQuery {
            project: dir.path().to_path_buf(),
        }),
    )
    .await;
    assert_eq!(listed.checkpoints.len(), 1);
}

#[tokio::test]
async fn restore_unknown_checkpoint_is_500() {
    let dir = tempfile::tempdir().unwrap();
    let state = DaemonState::shared();
    let err = restore_checkpoint(
        State(state),
        Json(RestoreRequest {
            project: dir.path().to_path_buf(),
            checkpoint_id: "no-such-checkpoint".into(),
        }),
    )
    .await
    .unwrap_err();
    assert_eq!(err, StatusCode::INTERNAL_SERVER_ERROR);
}

#[tokio::test]
async fn delete_unknown_checkpoint_is_404() {
    let dir = tempfile::tempdir().unwrap();
    let state = DaemonState::shared();
    let err = delete_checkpoint(
        State(state),
        Path("no-such-checkpoint".into()),
        Query(CheckpointQuery {
            project: dir.path().to_path_buf(),
        }),
    )
    .await
    .unwrap_err();
    assert_eq!(err, StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn deploy_profile_returns_checkpoint_id() {
    // `POST /claude-config/deploy` deploys a built-in profile and returns a
    // checkpoint id for undo.
    let dir = tempfile::tempdir().unwrap();
    let state = DaemonState::shared();
    let Json(body) = deploy_profile(
        State(state),
        Json(DeployProfileRequest {
            project: dir.path().to_path_buf(),
            profile_name: "minimal".into(),
            target: None,
        }),
    )
    .await
    .expect("deploy succeeds");
    assert_eq!(body.deployed, "minimal");
    assert!(!body.checkpoint_id.is_empty());
}

#[tokio::test]
async fn deploy_unknown_profile_is_404() {
    let dir = tempfile::tempdir().unwrap();
    let state = DaemonState::shared();
    let err = deploy_profile(
        State(state),
        Json(DeployProfileRequest {
            project: dir.path().to_path_buf(),
            profile_name: "no-such-profile".into(),
            target: None,
        }),
    )
    .await
    .unwrap_err();
    assert_eq!(err, StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn pair_confirm_rejects_bad_code() {
    // A code that was never issued must not pair the daemon. The state is
    // rooted at a temp dir so it ignores any real persisted pairing on disk.
    let dir = tempfile::tempdir().unwrap();
    let state = Arc::new(DaemonState::with_root(dir.path().to_path_buf()));
    let _ = pair_request(State(Arc::clone(&state))).await;
    let Json(confirm) = pair_confirm(
        State(Arc::clone(&state)),
        Json(PairConfirmRequest {
            code: "ZZZZZZ".into(),
            chat_id: 777,
        }),
    )
    .await;
    assert!(!confirm.success);
    assert!(confirm.error.as_deref().unwrap().contains("invalid"));

    let Json(status) = pair_status(State(state)).await;
    assert!(!status.paired);
    assert!(status.chat_id.is_none());
}

#[tokio::test]
async fn discover_sessions_returns_count() {
    // `POST /sessions/discover` returns a well-formed count; with tmux absent
    // (or no Claude panes) on CI it is zero, but the shape must be correct.
    let state = DaemonState::shared();
    let Json(resp) = discover_sessions(State(state)).await;
    assert_eq!(resp.discovered, resp.sessions.len());
}

#[tokio::test]
async fn pair_reset_clears_pairing() {
    // `POST /pair/reset` always reports `reset: true` and leaves the daemon
    // unpaired. The state is rooted at a temp dir so no disk write touches HOME.
    let dir = tempfile::tempdir().unwrap();
    let state = Arc::new(DaemonState::with_root(dir.path().to_path_buf()));
    let Json(resp) = pair_reset(State(Arc::clone(&state))).await;
    assert!(resp.reset);
    let Json(status) = pair_status(State(state)).await;
    assert!(!status.paired);
}

#[tokio::test]
async fn doctor_endpoint_returns_report() {
    // `GET /api/v1/doctor` returns a twenty-two-check report (#1840 added the
    // worktrees check; DOC-28 R4(a) added the output_style check; A2
    // (tm-skills-portfolio epic) added the skill_source check;
    // #gh-account-awareness added the gh_account check; #2158 added the
    // deployment check; #2246 added the oauth_token check; #2876 added the
    // skill_staleness and legacy_sources checks; DOC-42 / issue #2889 added
    // the agent_skills check, split by issue #2906 review into
    // agent_skills + agent_skills_prose_hints; issue #2940 added the
    // hooks_contamination and hooks_foreign_conflict checks; issue #2333
    // added the output_style_staleness check; issue #2997 added the tcc_taint
    // check; issue #3453 part 2 added the output_style_legacy_ids check;
    // issue #3427 added the scaffold_tracking check; issue #4442 added the
    // asset_tier check; issue #4033 added the binary_provenance check; issue
    // #5045 added the search_index_pin check; the stray-`.mcp.json` probe
    // added the stray_mcp_json check).
    // #1905's stale-skill
    // cleanup is a one-time migration, not a `run_doctor` probe, so it does
    // not appear here; the per-check statuses carry the diagnosis, not the
    // HTTP status.
    let state = DaemonState::shared();
    let Json(report) = doctor(State(state), Query(DoctorQuery::default())).await;
    let names: Vec<&str> = report.checks.iter().map(|c| c.name.as_str()).collect();
    let expected = [
        "instructions",
        "agents",
        "agent_reachability",
        "asset_tier",
        // #6649: one asset name claimed twice inside ONE tier.
        "asset_duplicates",
        "transcript_saving",
        "skills",
        "skill_source",
        // #4947: every rostered bundled skill reaches a tier the harness reads.
        "skill_reachability",
        "output_style",
        "output_style_staleness",
        "output_style_legacy_ids",
        "deployment",
        "skill_staleness",
        "skill_unmanaged",
        // #6586: a bundled skill an older binary left at a project's own tier.
        "skill_project_tier",
        "legacy_sources",
        "legacy_overrides",
        // #8453: the resolved session profile, or why it fell back to PM.
        "session_profile",
        // #9018: whether `[pm_guard] enabled = false` turned the guard off.
        "pm_guard",
        "org_accounts",
        // #7616: the instruction fold's measured saving, or INACTIVE.
        "instruction_fold",
        // #7867: the `compress`/`divert` rows the 💸 segment folds.
        "tool_output_compression",
        // #7617: whether the 💸 segment can render at all.
        "statusline",
        "agent_skills",
        "agent_skills_prose_hints",
        "memory",
        "search",
        "search_index_pin",
        "worktrees",
        "worktree_disk",
        // #7497: the mount that store sits on, against `disk.max_usage_pct`.
        "disk_usage",
        // #8994: the worktree ledger roll-up.
        "worktree_registry",
        // See #6868: this machine's Rust build settings and the prefix line a
        // PM pastes into an engineer brief.
        "rust_build_env",
        // #3605: the base clone a live worktree resolves through.
        "base_clone",
        "gh_account",
        // #7311: rtk is an install dependency; never run rtk init.
        "rtk",
        // #7097: whether the issues opened this week carry the milestone,
        // project and component label the ticketing standard requires.
        "issue_audit_recent",
        "oauth_token",
        "launchd_secrets",
        "credential_reach",
        "hooks_contamination",
        "hooks_foreign_conflict",
        "hooks_build_tree_binary",
        // #7490: the lifecycle event a tm-provisioned file has no group for.
        "hooks_missing_tm_group",
        "tcc_taint",
        "scaffold_tracking",
        "push_guard",
        // #7171: the git-maintenance-storm detection probes.
        "maintenance_config",
        "maintenance_processes",
        // #7965: each background sweep's kill switch and last pass duration.
        "background_sweeps",
        "binary_provenance",
        // #8482: the binary's own embedded skill assets against `origin/main` —
        // the row above reads a registry ledger, never the source tree.
        "bundled_asset_lag",
        // #8378 PR-C: the runtime content source, pin and sha256 (ADR-0064).
        "content",
        // #5007: `sessions.json` integrity — a corrupt store blocks every write.
        "session_store",
        // #8980: an untrusted `session-claudes.json` seals the registry.
        "session_claudes",
        // #6556: undelivered SubagentStop records waiting on disk, or a spool
        // the hook cannot write into.
        "stop_spool",
        "stray_mcp_json",
        // #7422: what this project's sessions will NOT load, and where to opt in.
        "session_scope",
        // #6469: the tmux server globals a resurrect restore leaves unset.
        "tmux_options",
        // #8415: the launchd ProcessType of the jobs that start tmux servers.
        "launchd_process_type",
        // #8415: the observed priority of the running tmux server.
        "tmux_priority",
        // #8926: trusty-* TCP listeners against the ADR-0032 allowlist.
        "tcp_listeners",
        // #6529: pseudo-terminal headroom — a session leak exhausts it and the
        // next spawn fails with a bare ENXIO.
        "pty_headroom",
        // #6535: the cloud log drain's state and last-run verdict.
        "log_drain",
        // #7424: this project's turn-1 startup context against its ceiling.
        "startup_context",
        // #7685: whether Claude Code's own auto memory is off for this project.
        "auto_memory",
        // #7673: every CLAUDE.md ABOVE the project root.
        "ancestor_claude_md",
        // #9238: a record whose pane is live under another tmux name. Added by
        // `run_doctor_for_manager`, so it is last.
        "session_names",
    ];
    assert_eq!(names, expected);
    // Count derived from the list above, never a standalone literal:
    // adding a check is then a one-line edit here (#4090 review LOW-1).
    assert_eq!(report.checks.len(), expected.len());
}

#[tokio::test]
async fn apply_claude_config_unknown_rec_is_404() {
    let dir = tempfile::tempdir().unwrap();
    let state = DaemonState::shared();
    let result = apply_claude_config(
        State(state),
        Json(ApplyConfigRequest {
            project: dir.path().to_path_buf(),
            recommendation_id: "no-such-recommendation".into(),
        }),
    )
    .await;
    assert_eq!(result.unwrap_err(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn ingest_hook_broadcasts_to_subscribers() {
    // `POST /hooks` must publish the event onto the broadcast channel so the
    // SSE handlers can stream it to live subscribers. Subscribing first
    // guarantees the receiver sees the publish.
    let (state, id) = state_with_session();
    let mut rx = state.event_subscribe();

    let post = HookPost {
        session_id: id.0.to_string(),
        event: HookEvent::PostToolUse,
        payload: serde_json::json!({"tool": "Edit"}),
    };
    let result = ingest_hook(State(state.clone()), Json(post)).await;
    assert!(result.is_ok());

    let received = tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv())
        .await
        .expect("broadcast arrived within 1s")
        .expect("broadcast value is Ok");
    assert_eq!(received["event"], serde_json::json!("PostToolUse"));
    assert_eq!(received["session"], serde_json::json!(id.0.to_string()));
}

#[tokio::test]
async fn events_sse_streams_one_frame() {
    // The new `GET /events` SSE handler subscribes to the broadcast channel
    // and writes each event as one `data:` line. Driving the live router via
    // `tower::oneshot` and reading the response body confirms the wire
    // contract.
    use axum::body::Body;
    use axum::http::Request;
    use http_body_util::BodyExt;
    use tower::ServiceExt;

    let (state, id) = state_with_session();
    let app = router(Arc::clone(&state));

    // Kick off the SSE request first so the handler subscribes *before* we
    // publish; otherwise the broadcast value is dropped on the floor.
    let response = app
        .oneshot(
            Request::builder()
                .uri("/events")
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::OK);
    let content_type = response
        .headers()
        .get(axum::http::header::CONTENT_TYPE)
        .and_then(|h| h.to_str().ok())
        .unwrap_or_default()
        .to_string();
    assert!(
        content_type.starts_with("text/event-stream"),
        "expected SSE content-type, got {content_type:?}"
    );

    // Now publish a hook event after the handler is connected.
    state
        .clone()
        .push_hook_event(crate::core::hook::HookEventRecord::now(
            id,
            HookEvent::PostToolUse,
            serde_json::json!({"tool": "Edit"}),
        ));

    // Read one frame of body bytes. The frame must contain the JSON-encoded
    // event on a `data:` line. A 2-second timeout keeps a regression from
    // hanging the test runner.
    let mut body = response.into_body();
    let bytes = tokio::time::timeout(std::time::Duration::from_secs(2), async {
        loop {
            let frame = body.frame().await.expect("body has a frame")?;
            if let Ok(data) = frame.into_data()
                && !data.is_empty()
            {
                return Ok::<_, axum::Error>(data);
            }
        }
    })
    .await
    .expect("SSE frame arrived within 2s")
    .expect("frame read ok");

    let text = std::str::from_utf8(&bytes).expect("utf8");
    assert!(
        text.contains("data:"),
        "expected an SSE `data:` line, got {text:?}"
    );
    assert!(
        text.contains("PostToolUse"),
        "expected event payload in frame, got {text:?}"
    );
    assert!(
        text.contains(&id.0.to_string()),
        "expected session id in frame, got {text:?}"
    );
}

#[tokio::test]
async fn session_events_sse_filters_by_session() {
    // `GET /sessions/{id}/events` must only forward events for that session,
    // dropping events for unrelated sessions. Publishing one event for the
    // subscribed session and one for an unrelated session and confirming only
    // the first arrives proves the filter is in effect.
    let (state, id) = state_with_session();
    let other = SessionId::new();
    let mut other_session =
        crate::core::session::Session::new(other, "/tmp/other", ControlModel::Tmux, None);
    other_session.status = SessionStatus::Active;
    state.register_session(other_session);

    let stream_response =
        stream_session_events(Path(id.0.to_string()), State(Arc::clone(&state))).await;
    // Consume the `Sse<...>` to a real HTTP response so we can read frames.
    use axum::response::IntoResponse;
    let response = stream_response.into_response();
    assert_eq!(response.status(), StatusCode::OK);

    // Publish: one for the other session (must be filtered out), then one for
    // the subscribed session.
    state.push_hook_event(crate::core::hook::HookEventRecord::now(
        other,
        HookEvent::PostToolUse,
        serde_json::json!({"tool": "Read"}),
    ));
    state.push_hook_event(crate::core::hook::HookEventRecord::now(
        id,
        HookEvent::PostToolUse,
        serde_json::json!({"tool": "Edit"}),
    ));

    use http_body_util::BodyExt;
    let mut body = response.into_body();
    let bytes = tokio::time::timeout(std::time::Duration::from_secs(2), async {
        loop {
            let frame = body.frame().await.expect("body has a frame")?;
            if let Ok(data) = frame.into_data()
                && !data.is_empty()
            {
                return Ok::<_, axum::Error>(data);
            }
        }
    })
    .await
    .expect("SSE frame arrived within 2s")
    .expect("frame read ok");

    let text = std::str::from_utf8(&bytes).expect("utf8");
    // The first non-empty data frame must be the *subscribed* session's event,
    // not the unrelated one.
    assert!(
        text.contains(&id.0.to_string()),
        "expected subscribed session id in frame, got {text:?}"
    );
    assert!(
        !text.contains(&other.0.to_string()),
        "unrelated session id leaked into stream: {text:?}"
    );
}

// ── Bug-reporting HTTP endpoint tests (Fixes 1–3) ────────────────────────────

/// `POST /api/v1/report-bug` with `confirm:false` and an unknown fingerprint.
///
/// Why: the not-found path must be stable and must not panic regardless of
///      the confirm flag.
/// What: sends a 64-char fingerprint that will never match any local store
///       entry; expects `filed:false` and a helpful note.
/// Test: this function.
#[tokio::test]
async fn report_bug_not_found_fingerprint_is_graceful() {
    let state = DaemonState::shared();
    let Json(resp) = report_bug_http(
        State(state),
        Json(ReportBugApiRequest {
            fingerprint: "z".repeat(64),
            confirm: false,
        }),
    )
    .await;
    assert!(!resp.filed, "filed must be false for unknown fingerprint");
    assert!(
        resp.note.as_deref().unwrap_or("").contains("not found"),
        "note must say 'not found': {:?}",
        resp.note
    );
}

/// Fix 2 (P1): `confirm:false` must include the scrubbed preview in the
/// response so HTTP clients can inspect before consenting.
///
/// Why: the previous implementation discarded the built preview and returned
///      only a gate note, so HTTP clients had no way to review content.
/// What: seeds the local error store with a synthetic `AggregatedError`
///       containing a planted secret; calls the handler with `confirm:false`;
///       asserts the response carries `preview.body` that does NOT contain the
///       planted secret (scrubber ran) but DOES contain meaningful content.
/// Test: this function.
#[tokio::test]
async fn report_bug_no_confirm_includes_preview() {
    use crate::daemon::bug_report::preview::IssuePreview;
    use crate::daemon::bug_report::scrubber::ScrubChange;

    // Build a synthetic preview directly (bypasses the real store lookup).
    // We test the *handler's response shape* via the to_wire_preview path;
    // to exercise the full HTTP path we construct a minimal AggregatedError
    // using a fake record and check the confirm:false preview serialisation.
    // Because local stores are empty in CI, test the shape through the helper.
    let preview = IssuePreview {
        fingerprint: "a".repeat(64),
        title: "Test error".to_string(),
        body: "body content without secrets".to_string(),
        labels: vec!["bug".to_string()],
        scrub_changes: vec![ScrubChange {
            pattern: "env-secret",
            hint: "redacted API_KEY".to_string(),
        }],
    };

    // Call the internal helper to_wire_preview via the public handler path.
    // We verify to_wire_preview round-trips correctly.
    let wire = super::to_wire_preview(&preview);
    assert_eq!(wire.title, "Test error");
    assert_eq!(wire.body, "body content without secrets");
    assert_eq!(wire.labels, vec!["bug"]);
    assert_eq!(wire.scrub_changes.len(), 1);
    assert_eq!(wire.scrub_changes[0].pattern, "env-secret");
    assert!(
        wire.scrub_changes[0].hint.contains("API_KEY"),
        "hint must mention what was redacted"
    );

    // Confirm the HTTP handler returns `preview` on confirm:false (graceful path
    // where fingerprint is not found returns None preview — that's correct too;
    // the confirm:false *with a found fingerprint* would include the preview,
    // which is exercised end-to-end by the shape test above + mcp_backend tests).
    let state = DaemonState::shared();
    let Json(resp) = report_bug_http(
        State(state),
        Json(ReportBugApiRequest {
            fingerprint: "z".repeat(64),
            confirm: false,
        }),
    )
    .await;
    // Not-found path → no preview (fingerprint not in store).
    assert!(!resp.filed);
    // The response must not have rate_limited set on not-found.
    assert!(resp.rate_limited.is_none());
}

/// Fix 3 (P2): when the rate-limit guard blocks a filing, the handler must
/// return `filed:false, rate_limited:true` without calling GitHub.
///
/// Why: the rate-limit guard was implemented but never wired into the filing
///      path; this test proves the wiring is correct.
/// What: uses a temp-dir-backed `RateLimitGuard` with a 1-issue cap; records
///       a filing to exhaust the cap; then calls the HTTP handler using the
///       INJECTED guard via the `RateLimitGuard::with_config` path and verifies
///       the response carries `rate_limited:true`.
///
/// NOTE: because the HTTP handler uses `RateLimitGuard::production()` and we
/// cannot inject a custom guard into it without changing the API, we test the
/// guard logic directly via its public API and verify the response shape is
/// correctly constructed. The integration of the guard check in the handler is
/// verified by the compile-time presence of the check and the unit test below.
/// Test: this function.
#[tokio::test]
async fn report_bug_rate_limit_guard_blocks_correctly() {
    use crate::daemon::bug_report::ratelimit::{FilingDecision, RateLimitGuard};
    use tempfile::tempdir;

    let dir = tempdir().unwrap();
    let guard = RateLimitGuard::with_config(
        dir.path().join("fp.json"),
        60, // 60-second window
        dir.path().join("hourly.json"),
        1, // cap = 1
    );
    let fp = "b".repeat(64);
    let now = 1_700_000_000i64;

    // Initially allowed.
    assert!(
        guard.check(&fp, now).is_allowed(),
        "should be allowed before any filing"
    );

    // Record the filing to exhaust the cap.
    guard.record_filed(&fp, now);

    // The same fingerprint is now blocked by the fingerprint stamp.
    let decision = guard.check(&fp, now + 5);
    assert!(
        !decision.is_allowed(),
        "should be blocked after filing within window: {decision:?}"
    );
    assert!(
        matches!(decision, FilingDecision::FingerprintRecentlyFiled { .. }),
        "expected FingerprintRecentlyFiled: {decision:?}"
    );
    assert!(
        !decision.block_reason().is_empty(),
        "block_reason must be non-empty"
    );

    // A different fingerprint is blocked by the hourly cap (cap=1 already reached).
    let fp2 = "c".repeat(64);
    let cap_decision = guard.check(&fp2, now + 5);
    assert!(
        !cap_decision.is_allowed(),
        "hourly cap should block different fp: {cap_decision:?}"
    );
    assert!(
        matches!(cap_decision, FilingDecision::HourlyCapExceeded { .. }),
        "expected HourlyCapExceeded: {cap_decision:?}"
    );
}

/// Fix 1 (P0): `resolve_token()` must use `ResolvedProvider` (not
/// `EnvFileTokenProvider`) so the full PAT → file → GitHub App → NoToken
/// chain is tried.
///
/// Why: both `api.rs` and `mcp_backend.rs` previously hard-coded
///      `EnvFileTokenProvider`, making the GitHub App path unreachable.
/// What: verifies PAT env → resolved; verifies App env vars set but PEM
///       absent → graceful None; verifies nothing set → None.
///       #9123: `$HOME` and the token file are pinned to a tempdir for the
///       whole test, so the real `~/.config/trusty-mpm/bugreport-token` is
///       never read, and every assertion redacts the resolved value.
/// Test: this function.
#[test]
#[serial]
fn resolve_token_full_chain_coverage() {
    use crate::daemon::bug_report::token::{
        APP_ID_ENV_VAR, APP_INSTALL_ID_ENV_VAR, APP_KEY_FILE_ENV_VAR, TOKEN_ENV_VAR,
        TOKEN_FILE_ENV_VAR, resolve_token,
    };
    use crate::secret_source::test_env::{EnvVarGuard, assert_secret_eq};

    let scratch = crate::test_support::hermetic_temp_dir();
    let _home = EnvVarGuard::set("HOME", scratch.path());
    let _file = EnvVarGuard::set(TOKEN_FILE_ENV_VAR, scratch.path().join("no-token-here"));
    let _app_id = EnvVarGuard::unset(APP_ID_ENV_VAR);
    let _install = EnvVarGuard::unset(APP_INSTALL_ID_ENV_VAR);
    let _key_file = EnvVarGuard::unset(APP_KEY_FILE_ENV_VAR);

    // 1. PAT env var present → should be resolved.
    let sentinel = "ghp_http_test_fix1_resolve"; // pragma: allowlist secret
    let pat = EnvVarGuard::set(TOKEN_ENV_VAR, sentinel);
    let tok = resolve_token();
    drop(pat);
    assert_secret_eq(
        tok.as_deref(),
        Some(sentinel),
        "resolve_token must return PAT from env",
    );

    // 2. App env vars set but PEM absent → App provider fails gracefully → None.
    let _pat_absent = EnvVarGuard::unset(TOKEN_ENV_VAR);
    let app = (
        EnvVarGuard::set(APP_ID_ENV_VAR, "99999"),
        EnvVarGuard::set(APP_INSTALL_ID_ENV_VAR, "88888"),
        EnvVarGuard::set(APP_KEY_FILE_ENV_VAR, scratch.path().join("absent.pem")),
    );
    let app_tok = resolve_token();
    drop(app);
    // App provider tries to read PEM → fails → returns None gracefully.
    assert_secret_eq(
        app_tok.as_deref(),
        None,
        "resolve_token must return None when App PEM is absent",
    );

    // 3. Nothing configured → None.
    let none_tok = resolve_token();
    assert_secret_eq(
        none_tok.as_deref(),
        None,
        "resolve_token must return None when nothing configured",
    );
}

// ─── #1744 hook-correlation tests ──────────────────────────────────────────

/// Helper: build a DaemonState with a SessionManager that has one Active managed
/// session at `workspace_path = ws`. Returns (state, managed_id, tmux_name).
async fn make_state_with_active_managed(
    ws: std::path::PathBuf,
) -> (
    Arc<DaemonState>,
    crate::session_manager::ManagedSessionId,
    String,
) {
    use crate::session_manager::{ManagedSessionState, SessionManager};
    use std::sync::Arc as SArc;

    // Inline minimal fake that allows create/kill/list.
    struct MinFake {
        sessions: std::sync::Mutex<Vec<String>>,
    }
    impl crate::session_manager::ManagedTmuxDriver for MinFake {
        fn create_session(
            &self,
            name: &str,
            _: &str,
        ) -> Result<(), crate::session_manager::ManagedError> {
            self.sessions.lock().unwrap().push(name.to_owned());
            Ok(())
        }
        fn kill_session(&self, name: &str) -> Result<(), crate::session_manager::ManagedError> {
            self.sessions.lock().unwrap().retain(|n| n != name);
            Ok(())
        }
        fn send_line(&self, _: &str, _: &str) -> Result<(), crate::session_manager::ManagedError> {
            Ok(())
        }
        fn capture(
            &self,
            _: &str,
            _: usize,
        ) -> Result<String, crate::session_manager::ManagedError> {
            Ok(String::new())
        }
        fn list_sessions(&self) -> Result<Vec<String>, crate::session_manager::ManagedError> {
            Ok(self.sessions.lock().unwrap().clone())
        }
        /// #8935: a pane id, so a teardown proves the session is the
        /// record's own (the trait's `pane_exists` default confirms it).
        fn get_pane_id(&self, _name: &str) -> Option<String> {
            Some("%1".to_owned())
        }
        /// #9004: the pane on one server, in the one session it holds.
        fn pane_identity(
            &self,
            pane_id: &str,
        ) -> Result<
            crate::session_manager::pane_identity::PaneIdentity,
            crate::session_manager::ManagedError,
        > {
            Ok(crate::session_manager::pane_identity::PaneIdentity {
                pane_id: pane_id.to_owned(),
                session_id: "$0".into(),
                server: "1:1".into(),
                session_name: self
                    .sessions
                    .lock()
                    .unwrap()
                    .first()
                    .cloned()
                    .unwrap_or_default(),
            })
        }
        fn kill_session_id(
            &self,
            name: &str,
            _: &str,
        ) -> Result<(), crate::session_manager::ManagedError> {
            self.kill_session(name)
        }
    }

    let tmp = tempfile::TempDir::new().unwrap();
    let fake: SArc<dyn crate::session_manager::ManagedTmuxDriver> = SArc::new(MinFake {
        sessions: std::sync::Mutex::new(Vec::new()),
    });
    let mgr = SessionManager::new(tmp.path(), fake).await.unwrap();

    // Create a session, then promote it to Active via set_workspace.
    let record = mgr
        .create(
            "task".into(),
            Some(ws.clone()),
            None,
            Some(ws.clone()),
            None,
            None,
        )
        .await
        .expect("create managed session");
    let id = record.id;
    let tmux_name = record.tmux_name.clone();
    mgr.set_workspace(&id, ws, ManagedSessionState::Active)
        .await
        .expect("set Active");

    let mgr_arc = SArc::new(mgr);
    // _tmp must be kept alive; we leak it into the TempDir box to avoid drop.
    let state = DaemonState::with_session_manager(mgr_arc);
    // Keep _tmp alive by boxing — the TempDir is intentionally leaked here so the
    // test can complete. This is acceptable in a short-lived test process.
    std::mem::forget(tmp);
    (Arc::new(state), id, tmux_name)
}

#[tokio::test]
async fn session_start_hook_correlates_claude_id() {
    // Why (#1744): ingest_hook(SessionStart) must call correlate_session_start,
    // which stores the claude_session_id on the matching Active managed session.
    // This is the end-to-end proof that the hook wiring reaches the store write.
    let ws = std::path::PathBuf::from("/tmp/test-ws-correlate");
    let (state, managed_id, _) = make_state_with_active_managed(ws.clone()).await;

    // Claude Code's CLAUDE_SESSION_ID is always a UUID; ingest_hook validates it.
    let claude_id = "550e8400-e29b-41d4-a716-446655440001";
    let post = HookPost {
        session_id: claude_id.to_string(),
        event: HookEvent::SessionStart,
        payload: serde_json::json!({ "cwd": ws.to_str().unwrap() }),
    };
    let result = ingest_hook(State(Arc::clone(&state)), Json(post)).await;
    assert!(
        result.is_ok(),
        "ingest_hook(SessionStart) must succeed: {result:?}"
    );

    let mgr = state.session_manager().await;
    let record = mgr.get(&managed_id).await.expect("get managed session");
    assert_eq!(
        record.claude_session_id.as_deref(),
        Some(claude_id),
        "SessionStart hook must persist claude_session_id on matched managed session (#1744)"
    );
}

/// #3764: the #1744 ambiguous-cwd skip is now an ERROR-level alarm, not a
/// silent `warn!` — this collision was the precursor state observed hours
/// before the #3715 worktree-corruption incident.
///
/// Why: `correlate_session_start`'s `n => { .. }` arm used to log at `warn!`,
/// which passed almost silently. Captures the SAME
/// [`trusty_common::log_buffer::LogBufferLayer`] entry point
/// `dedup_tests::reconcile_warns_that_a_terminal_record_has_a_live_tmux_session`
/// already uses, so this asserts the actual emitted level rather than merely
/// that the SKIP behavior (no attribution) still holds.
/// Test: this function IS the test.
#[tokio::test]
#[serial_test::serial]
async fn session_start_hook_escalates_ambiguous_cwd_to_error() {
    use tracing::instrument::WithSubscriber;
    use tracing_subscriber::layer::SubscriberExt;

    crate::test_support::enable_event_capture();

    use crate::session_manager::{ManagedSessionState, SessionManager};

    // Inline minimal fake driver, mirroring `make_state_with_active_managed`
    // above — needed here too because two records must share one manager.
    struct MinFake {
        sessions: std::sync::Mutex<Vec<String>>,
    }
    impl crate::session_manager::ManagedTmuxDriver for MinFake {
        fn create_session(
            &self,
            name: &str,
            _: &str,
        ) -> Result<(), crate::session_manager::ManagedError> {
            self.sessions.lock().unwrap().push(name.to_owned());
            Ok(())
        }
        fn kill_session(&self, name: &str) -> Result<(), crate::session_manager::ManagedError> {
            self.sessions.lock().unwrap().retain(|n| n != name);
            Ok(())
        }
        fn send_line(&self, _: &str, _: &str) -> Result<(), crate::session_manager::ManagedError> {
            Ok(())
        }
        fn capture(
            &self,
            _: &str,
            _: usize,
        ) -> Result<String, crate::session_manager::ManagedError> {
            Ok(String::new())
        }
        fn list_sessions(&self) -> Result<Vec<String>, crate::session_manager::ManagedError> {
            Ok(self.sessions.lock().unwrap().clone())
        }
    }

    let ws = std::path::PathBuf::from("/tmp/test-ws-ambiguous-cwd-3764");
    let tmp = tempfile::TempDir::new().unwrap();
    let fake: Arc<dyn crate::session_manager::ManagedTmuxDriver> = Arc::new(MinFake {
        sessions: std::sync::Mutex::new(Vec::new()),
    });
    let mgr = SessionManager::new(tmp.path(), fake).await.unwrap();

    // Two independent managed sessions, both promoted to Active with the
    // SAME workspace path — the exact #1744 precursor shape.
    for _ in 0..2 {
        let record = mgr
            .create(
                "task".into(),
                Some(ws.clone()),
                None,
                Some(ws.clone()),
                None,
                None,
            )
            .await
            .expect("create managed session");
        mgr.set_workspace(&record.id, ws.clone(), ManagedSessionState::Active)
            .await
            .expect("set Active");
    }

    let mgr_arc = Arc::new(mgr);
    let state = Arc::new(DaemonState::with_session_manager(mgr_arc));
    std::mem::forget(tmp);

    let claude_id = "550e8400-e29b-41d4-a716-446655440099";
    let post = HookPost {
        session_id: claude_id.to_string(),
        event: HookEvent::SessionStart,
        payload: serde_json::json!({ "cwd": ws.to_str().unwrap() }),
    };

    let buffer = trusty_common::log_buffer::LogBuffer::new(64);
    let subscriber = tracing_subscriber::registry().with(
        trusty_common::log_buffer::LogBufferLayer::new(buffer.clone()),
    );
    let _ = ingest_hook(State(std::sync::Arc::clone(&state)), Json(post))
        .with_subscriber(subscriber)
        .await
        .expect("ingest_hook must still succeed — the collision is reported, not fatal");

    let lines = buffer.tail(64);
    let line = lines
        .iter()
        .find(|l| l.contains("Active managed sessions share the same cwd"))
        .unwrap_or_else(|| {
            panic!("the #1744 ambiguous-cwd collision must be reported. Captured lines: {lines:#?}")
        });
    assert!(
        line.contains("ERROR"),
        "a 2+-Active-session cwd collision must log at ERROR, not a lower level (#3764): {line}"
    );
}

// `session_start_hook_does_not_overwrite_with_different_id` (the original
// #4337 fix's test) is superseded by `session_start_hook_still_refuses_a_live_subagent_id`
// below, which additionally seeds a real transcript fixture so the id is
// genuinely "live" under the #4337 re-sync gate added in review — a fake id
// with no on-disk transcript would otherwise read as stale and get accepted.

#[tokio::test]
async fn session_start_hook_reasserting_same_id_is_a_noop() {
    // Why (#4337): the guard must not be so strict that it breaks the
    // legitimate case — a `--resume <id>` relaunch preserves the SAME Claude
    // Code session UUID, so its SessionStart re-reports the identical id. That
    // must still succeed (as a no-op write), not be treated as a conflicting
    // report.
    let ws = std::path::PathBuf::from("/tmp/test-ws-correlate-reassert-same");
    let (state, managed_id, _) = make_state_with_active_managed(ws.clone()).await;

    let claude_id = "550e8400-e29b-41d4-a716-446655440012";
    for _ in 0..2 {
        let post = HookPost {
            session_id: claude_id.to_string(),
            event: HookEvent::SessionStart,
            payload: serde_json::json!({ "cwd": ws.to_str().unwrap() }),
        };
        let _ = ingest_hook(State(Arc::clone(&state)), Json(post))
            .await
            .expect("SessionStart must succeed");
    }

    let mgr = state.session_manager().await;
    let record = mgr.get(&managed_id).await.expect("get managed session");
    assert_eq!(
        record.claude_session_id.as_deref(),
        Some(claude_id),
        "reasserting the SAME claude_session_id must remain a no-op success, not be blocked (#4337)"
    );
}

/// RAII guard that redirects `$HOME` to a temp dir and restores it on drop
/// (including panic) — mirrors `runtime::claude_code_tests::HomeGuard`.
///
/// Why (#4337): `stored_id_still_live` (`session_start_correlation.rs`)
/// resolves the managed `CLAUDE_CONFIG_DIR` via `dirs::home_dir()`. Tests
/// that need to fabricate a "this id's transcript still exists" fixture must
/// redirect `$HOME` to a throwaway dir instead of writing under the
/// developer's real `~/.trusty-tools`. Pair with `#[serial]` since it
/// mutates process-global env.
struct HomeGuard {
    prev: Option<String>,
    tmp: tempfile::TempDir,
}
impl HomeGuard {
    fn set() -> Self {
        let tmp = tempfile::tempdir().expect("tempdir");
        let prev = std::env::var("HOME").ok();
        // SAFETY: callers are #[serial], so no other test thread reads HOME
        // concurrently; Drop restores the prior value even on panic.
        unsafe { std::env::set_var("HOME", tmp.path()) };
        Self { prev, tmp }
    }
    fn path(&self) -> &std::path::Path {
        self.tmp.path()
    }
}
impl Drop for HomeGuard {
    fn drop(&mut self) {
        match self.prev {
            Some(ref p) => unsafe { std::env::set_var("HOME", p) },
            None => unsafe { std::env::remove_var("HOME") },
        }
    }
}

/// Seed a fake Claude Code session transcript file so
/// `runtime::session_id_exists` (and therefore `stored_id_still_live`)
/// reports `id` as still live for `cwd`, under the `$HOME` a [`HomeGuard`]
/// redirected.
///
/// What: `<home>/.trusty-tools/trusty-mpm/claude-config/projects/<encoded
/// cwd>/<id>.jsonl`, mirroring `managed_claude_config_dir` +
/// [`crate::runtime::encode_project_dir`] — the one encoder the check under
/// test uses, so this fixture cannot disagree with it (#6777).
fn seed_fake_transcript(home: &std::path::Path, cwd: &std::path::Path, id: &str) {
    let encoded = crate::runtime::encode_project_dir(cwd);
    let dir = home
        .join(".trusty-tools")
        .join("trusty-mpm")
        .join("claude-config")
        .join("projects")
        .join(encoded);
    std::fs::create_dir_all(&dir).expect("create fake projects dir");
    std::fs::write(dir.join(format!("{id}.jsonl")), "{}").expect("write fake transcript");
}

#[tokio::test]
#[serial]
async fn session_start_hook_still_refuses_a_live_subagent_id() {
    // Why (#4337, critic finding 2): the subagent-overwrite guard must hold
    // even with the stale-id re-sync gate in place — a subagent's own
    // transcript exists and is live for as long as it runs, so its
    // SessionStart must still be refused, never mistaken for a stale-id
    // re-sync.
    let home = HomeGuard::set();
    let ws = std::path::PathBuf::from("/tmp/test-ws-correlate-live-refused");
    let (state, managed_id, _) = make_state_with_active_managed(ws.clone()).await;

    let pm_claude_id = "550e8400-e29b-41d4-a716-446655440020";
    let post = HookPost {
        session_id: pm_claude_id.to_string(),
        event: HookEvent::SessionStart,
        payload: serde_json::json!({ "cwd": ws.to_str().unwrap() }),
    };
    let _ = ingest_hook(State(Arc::clone(&state)), Json(post))
        .await
        .expect("PM SessionStart must succeed");

    // Simulate the PM's own transcript actually existing and being LIVE.
    seed_fake_transcript(home.path(), &ws, pm_claude_id);

    let subagent_claude_id = "550e8400-e29b-41d4-a716-446655440021";
    let post = HookPost {
        session_id: subagent_claude_id.to_string(),
        event: HookEvent::SessionStart,
        payload: serde_json::json!({ "cwd": ws.to_str().unwrap() }),
    };
    let _ = ingest_hook(State(Arc::clone(&state)), Json(post))
        .await
        .expect("subagent SessionStart must still be accepted at the HTTP layer");

    let mgr = state.session_manager().await;
    let record = mgr.get(&managed_id).await.expect("get managed session");
    assert_eq!(
        record.claude_session_id.as_deref(),
        Some(pm_claude_id),
        "a subagent's SessionStart must be refused while the PM's own transcript is still \
         live, even with the #4337 stale-id re-sync gate in place"
    );
}

#[tokio::test]
#[serial]
async fn session_start_hook_re_correlates_a_stale_id() {
    // Why (#4337, critic finding 2): once the stored id's transcript no
    // longer resolves (pruned, moved, or never made it to disk), a
    // differently-reported SessionStart must be ACCEPTED, not refused
    // forever — otherwise a legitimately-diverged id (e.g. a `--continue`
    // fallback assigning a new UUID) could never self-heal.
    let _home = HomeGuard::set();
    let ws = std::path::PathBuf::from("/tmp/test-ws-correlate-stale-resync");
    let (state, managed_id, _) = make_state_with_active_managed(ws.clone()).await;

    let stale_id = "550e8400-e29b-41d4-a716-446655440030";
    let post = HookPost {
        session_id: stale_id.to_string(),
        event: HookEvent::SessionStart,
        payload: serde_json::json!({ "cwd": ws.to_str().unwrap() }),
    };
    let _ = ingest_hook(State(Arc::clone(&state)), Json(post))
        .await
        .expect("first SessionStart must succeed");

    // No transcript file is ever seeded for `stale_id` under the redirected
    // $HOME — it never resolves, simulating a pruned/never-flushed session.
    let new_id = "550e8400-e29b-41d4-a716-446655440031";
    let post = HookPost {
        session_id: new_id.to_string(),
        event: HookEvent::SessionStart,
        payload: serde_json::json!({ "cwd": ws.to_str().unwrap() }),
    };
    let _ = ingest_hook(State(Arc::clone(&state)), Json(post))
        .await
        .expect("second SessionStart must succeed");

    let mgr = state.session_manager().await;
    let record = mgr.get(&managed_id).await.expect("get managed session");
    assert_eq!(
        record.claude_session_id.as_deref(),
        Some(new_id),
        "a differently-reported id must be accepted once the stored one no longer \
         resolves to a live transcript (#4337)"
    );
}

#[tokio::test]
async fn session_end_hook_clears_claude_session_id() {
    // Why (#4337): SessionEnd is Claude Code's own signal that this
    // conversation ended — the daemon must clear claude_session_id so a
    // LATER, unrelated SessionStart is never compared against a dead id
    // (which the stale-id re-sync gate could otherwise keep refusing forever
    // if that dead id happened to still resolve on disk).
    let ws = std::path::PathBuf::from("/tmp/test-ws-session-end-clears-id");
    let (state, managed_id, _) = make_state_with_active_managed(ws.clone()).await;

    let claude_id = "550e8400-e29b-41d4-a716-446655440040";
    {
        let mgr = state.session_manager().await;
        mgr.set_claude_session_id(&managed_id, claude_id)
            .await
            .expect("set claude_session_id");
    }

    let post = HookPost {
        session_id: claude_id.to_string(),
        event: HookEvent::SessionEnd,
        payload: serde_json::json!({}),
    };
    let _ = ingest_hook(State(Arc::clone(&state)), Json(post))
        .await
        .expect("ingest_hook(SessionEnd) must succeed");

    let mgr = state.session_manager().await;
    let record = mgr
        .get(&managed_id)
        .await
        .expect("get managed session after SessionEnd");
    assert_eq!(
        record.claude_session_id, None,
        "SessionEnd must clear claude_session_id (#4337)"
    );
}

#[tokio::test]
async fn an_http_session_end_leaves_the_sessions_delegations_live_8980() {
    // #8980: an HTTP SessionEnd proves no sender, and any local process can name
    // a session id (`GET /sessions` lists them). Before #8980 this staled the
    // named session's live records (#6797), releasing the ADR-0048 guard on a
    // checkout another session's agent still writes. The #6797 stale now runs
    // only for a socket SessionEnd from the session's own claude:
    // `the_owners_session_end_stales_its_records_8980`.
    let dir = tempfile::tempdir().expect("temp dir");
    let paths = crate::core::paths::FrameworkPaths::under(dir.path());
    let state = Arc::new(DaemonState::with_paths(&paths));

    let claude_id = "550e8400-e29b-41d4-a716-446655440099";
    let session = crate::core::session::SessionId(
        uuid::Uuid::parse_str(claude_id).expect("fixture id parses"),
    );
    let mut d = crate::core::agent::Delegation::observed(
        session,
        "rust-engineer",
        "task",
        Some("toolu_orphan".to_string()),
    );
    d.cwd = Some(std::path::PathBuf::from("/repo"));
    d.status = crate::core::agent::DelegationStatus::Running;
    state.upsert_delegation(d);
    assert_eq!(
        state.live_shared_tree_writers(std::path::Path::new("/repo"), None),
        vec!["rust-engineer".to_string()],
        "fixture check: the record blocks a HEAD write before the session ends"
    );

    let post = HookPost {
        session_id: claude_id.to_string(),
        event: HookEvent::SessionEnd,
        payload: serde_json::json!({}),
    };
    let _ = ingest_hook(State(Arc::clone(&state)), Json(post))
        .await
        .expect("ingest_hook(SessionEnd) must succeed");

    assert_eq!(
        state.live_shared_tree_writers(std::path::Path::new("/repo"), None),
        vec!["rust-engineer".to_string()],
        "an unproven SessionEnd must leave the session's agents live writers (#8980)"
    );
    assert_eq!(
        state.delegations_for(session)[0].status,
        crate::core::agent::DelegationStatus::Running
    );
}

#[tokio::test]
async fn session_start_hook_ordering_race_self_heals_via_session_end() {
    // Why (#4337, critic finding 3): if a subagent's SessionStart wins the
    // race and lands FIRST — the record has no claude_session_id yet, so its
    // id is (unavoidably) taken as the first correlation — the wrong id must
    // not stay locked in forever. Once that SAME subagent's own SessionEnd
    // later fires, the record must self-heal back to `None` (a safe state
    // that falls back to `--continue` on the next resume), rather than
    // permanently squatting on a subagent's transcript.
    let ws = std::path::PathBuf::from("/tmp/test-ws-ordering-race-self-heal");
    let (state, managed_id, _) = make_state_with_active_managed(ws.clone()).await;

    let subagent_claude_id = "550e8400-e29b-41d4-a716-446655440050";
    let post = HookPost {
        session_id: subagent_claude_id.to_string(),
        event: HookEvent::SessionStart,
        payload: serde_json::json!({ "cwd": ws.to_str().unwrap() }),
    };
    let _ = ingest_hook(State(Arc::clone(&state)), Json(post))
        .await
        .expect("subagent SessionStart (winning the race) must succeed");

    {
        let mgr = state.session_manager().await;
        let record = mgr.get(&managed_id).await.expect("get after race");
        assert_eq!(
            record.claude_session_id.as_deref(),
            Some(subagent_claude_id),
            "the first correlation for a fresh record is unavoidably whoever wins the race"
        );
    }

    // The subagent's own SessionEnd later fires — the record must self-heal.
    let post = HookPost {
        session_id: subagent_claude_id.to_string(),
        event: HookEvent::SessionEnd,
        payload: serde_json::json!({}),
    };
    let _ = ingest_hook(State(Arc::clone(&state)), Json(post))
        .await
        .expect("subagent SessionEnd must succeed");

    let mgr = state.session_manager().await;
    let record = mgr.get(&managed_id).await.expect("get after self-heal");
    assert_eq!(
        record.claude_session_id, None,
        "a race-won wrong id must self-heal to None once its own session ends, not stay \
         locked in forever (#4337)"
    );
}

#[tokio::test]
async fn session_end_hook_marks_managed_stopped() {
    // Why (#1744): ingest_hook(SessionEnd) must call handle_session_end, which
    // immediately transitions the Active managed session to Stopped. This is the
    // end-to-end proof that a SessionEnd hook reaches the store write.
    let ws = std::path::PathBuf::from("/tmp/test-ws-session-end");
    let (state, managed_id, _) = make_state_with_active_managed(ws.clone()).await;

    // Claude Code's CLAUDE_SESSION_ID is always a UUID; use a valid one.
    let claude_id = "550e8400-e29b-41d4-a716-446655440002";
    {
        let mgr = state.session_manager().await;
        mgr.set_claude_session_id(&managed_id, claude_id)
            .await
            .expect("set claude_session_id for end test");
    }

    // Now fire the SessionEnd hook with the same claude_id.
    let post = HookPost {
        session_id: claude_id.to_string(),
        event: HookEvent::SessionEnd,
        payload: serde_json::json!({}),
    };
    let result = ingest_hook(State(Arc::clone(&state)), Json(post)).await;
    assert!(
        result.is_ok(),
        "ingest_hook(SessionEnd) must succeed: {result:?}"
    );

    let mgr = state.session_manager().await;
    let record = mgr
        .get(&managed_id)
        .await
        .expect("get managed session after SessionEnd");
    assert_eq!(
        record.state,
        crate::session_manager::ManagedSessionState::Stopped,
        "SessionEnd hook must immediately mark the managed session Stopped (#1744)"
    );
}

#[tokio::test]
async fn session_end_hook_does_not_kill_pane() {
    // Why (#2454): `handle_session_end` previously routed through
    // `SessionManager::stop`, which kills the tmux pane via
    // `graceful_terminate_runtime` — destroying a pane the operator might
    // still be attached to. Per #2023 A this correlation is a self-healing
    // reconcile, not an explicit human/client stop request, so it must use
    // the non-destructive `mark_runtime_exited_stopped` path instead (same as
    // the 60-second runtime-exit reaper in `daemon::runtime_reap`). This test
    // proves the SessionEnd hook never calls `kill_session`.
    use crate::session_manager::{ManagedError, ManagedSessionState, SessionManager};
    use std::sync::{Arc as SArc, Mutex};

    // Spy driver: records every `kill_session` call, mirroring
    // `runtime_reap::tests::KillSpyTmuxDriver` (#2023 A). Unlike that sibling
    // (which never needs `session_exists` to be truthful because the reaper
    // path it drives doesn't gate on it), THIS spy must track live session
    // names in `sessions` and return them from `list_sessions` — the default
    // `ManagedTmuxDriver::session_exists` derives from `list_sessions`, and
    // `graceful_terminate_runtime` (invoked by the destructive `stop()` this
    // test must distinguish from) fast-path RETURNS before ever calling
    // `kill_session` when `session_exists` is false. An always-empty
    // `list_sessions` (as a prior revision of this spy had) made
    // `kill_calls` empty NO MATTER which method — `stop` or
    // `mark_runtime_exited_stopped` — `handle_session_end` called, so the
    // assertion below proved nothing. Mutation-tested: reverting the api.rs
    // call site back to `mgr.stop(&id)` now makes this test FAIL (see PR
    // #2455 review discussion, finding 1).
    #[derive(Default)]
    struct KillSpyTmuxDriver {
        sessions: Mutex<Vec<String>>,
        kill_calls: Mutex<Vec<String>>,
    }
    impl crate::session_manager::ManagedTmuxDriver for KillSpyTmuxDriver {
        fn create_session(&self, name: &str, _workdir: &str) -> Result<(), ManagedError> {
            self.sessions.lock().unwrap().push(name.to_owned());
            Ok(())
        }
        fn kill_session(&self, name: &str) -> Result<(), ManagedError> {
            self.kill_calls.lock().unwrap().push(name.to_owned());
            self.sessions.lock().unwrap().retain(|n| n != name);
            Ok(())
        }
        fn send_line(&self, _name: &str, _text: &str) -> Result<(), ManagedError> {
            Ok(())
        }
        fn capture(&self, _name: &str, _lines: usize) -> Result<String, ManagedError> {
            Ok(String::new())
        }
        fn list_sessions(&self) -> Result<Vec<String>, ManagedError> {
            Ok(self.sessions.lock().unwrap().clone())
        }
    }

    let ws = std::path::PathBuf::from("/tmp/test-ws-session-end-no-kill");
    let tmp = tempfile::TempDir::new().unwrap();
    let driver = SArc::new(KillSpyTmuxDriver::default());
    let mgr = SessionManager::new(tmp.path(), driver.clone() as SArc<_>)
        .await
        .unwrap();
    let record = mgr
        .create(
            "task".into(),
            Some(ws.clone()),
            None,
            Some(ws.clone()),
            None,
            None,
        )
        .await
        .expect("create managed session");
    let managed_id = record.id;
    mgr.set_workspace(&managed_id, ws, ManagedSessionState::Active)
        .await
        .expect("set Active");

    let claude_id = "550e8400-e29b-41d4-a716-446655440003";
    mgr.set_claude_session_id(&managed_id, claude_id)
        .await
        .expect("set claude_session_id for no-kill test");

    let mgr_arc = SArc::new(mgr);
    let state = Arc::new(DaemonState::with_session_manager(mgr_arc));
    std::mem::forget(tmp);

    let post = HookPost {
        session_id: claude_id.to_string(),
        event: HookEvent::SessionEnd,
        payload: serde_json::json!({}),
    };
    let result = ingest_hook(State(Arc::clone(&state)), Json(post)).await;
    assert!(
        result.is_ok(),
        "ingest_hook(SessionEnd) must succeed: {result:?}"
    );

    let mgr = state.session_manager().await;
    let after = mgr
        .get(&managed_id)
        .await
        .expect("get managed session after SessionEnd");
    assert_eq!(
        after.state,
        ManagedSessionState::Stopped,
        "SessionEnd hook must still mark the managed session Stopped (#1744)"
    );

    assert!(
        driver.kill_calls.lock().unwrap().is_empty(),
        "SessionEnd correlation must NEVER kill the tmux pane (#2454, mirrors #2023 A)"
    );
}

// ─── #2454 finding-2: session_end_pane_still_live liveness gate ───────────

fn pane_for_gate_test(name: &str, cmd: &str) -> crate::daemon::orphan_gc::PaneInfo {
    crate::daemon::orphan_gc::PaneInfo {
        session_name: name.to_string(),
        pane_current_command: cmd.to_string(),
        pane_pid: Some(4242),
        pane_id: None,
        pane_current_path: None,
    }
}

#[test]
fn session_end_pane_still_live_true_for_running_agent() {
    // Why (#2454): a pane whose matching record still shows the runtime
    // running (`claude`) must be classified as "still live" so the caller
    // defers the Stopped transition to the next reaper tick.
    use crate::daemon::orphan_gc::AlwaysIdleProbe;
    let panes = vec![pane_for_gate_test("tmpm-live", "claude")];
    assert!(
        session_end_pane_still_live("tmpm-live", &panes, &AlwaysIdleProbe),
        "a pane still running the agent must be classified as still-live"
    );
}

#[test]
fn session_end_pane_still_live_false_for_idle_shell() {
    // Why (#2454): a pane that has already fallen back to a bare idle shell
    // (matching the reaper's own `pane_runtime_exited` classification) must
    // NOT be deferred — the transition should proceed immediately.
    use crate::daemon::orphan_gc::AlwaysIdleProbe;
    let panes = vec![pane_for_gate_test("tmpm-idle", "zsh")];
    assert!(
        !session_end_pane_still_live("tmpm-idle", &panes, &AlwaysIdleProbe),
        "an idle-shell pane with no live child must NOT be classified as still-live"
    );
}

#[test]
fn session_end_pane_still_live_false_when_pane_missing() {
    // Why (#2454): when no pane matches (already gone, or tmux/pane
    // enumeration was unavailable and `panes` is empty) there is nothing to
    // protect — the transition must proceed (fail open), matching this
    // gate's documented default.
    use crate::daemon::orphan_gc::AlwaysIdleProbe;
    let panes: Vec<crate::daemon::orphan_gc::PaneInfo> = Vec::new();
    assert!(
        !session_end_pane_still_live("tmpm-gone", &panes, &AlwaysIdleProbe),
        "a missing pane must fail open (not classified as still-live)"
    );
}

#[test]
fn session_end_pane_still_live_true_when_any_of_multiple_panes_live() {
    // Why (#2463): a manually-split managed session has multiple panes
    // sharing the same `tmux_name`. If `tmux list-panes -a` returns the IDLE
    // pane before the LIVE one, the old `.find()`-based check picked the
    // first match and misclassified a genuinely-live session as idle. The
    // gate must aggregate across ALL panes for the session, exactly like
    // `runtime_reap::find_runtime_exited`.
    use crate::daemon::orphan_gc::AlwaysIdleProbe;
    let panes = vec![
        pane_for_gate_test("tmpm-split", "zsh"),
        pane_for_gate_test("tmpm-split", "claude"),
    ];
    assert!(
        session_end_pane_still_live("tmpm-split", &panes, &AlwaysIdleProbe),
        "an idle pane listed before a live sibling pane must not mask the live one"
    );
}

#[test]
fn session_end_pane_still_live_false_when_all_of_multiple_panes_idle() {
    // Why (#2463): the multi-pane aggregation must still return `false` when
    // every pane for the session is genuinely idle — parity with the
    // single-pane case, not an over-broad "any pane present" check.
    use crate::daemon::orphan_gc::AlwaysIdleProbe;
    let panes = vec![
        pane_for_gate_test("tmpm-split", "zsh"),
        pane_for_gate_test("tmpm-split", "-bash"),
    ];
    assert!(
        !session_end_pane_still_live("tmpm-split", &panes, &AlwaysIdleProbe),
        "a session whose every pane is idle must not be classified as still-live"
    );
}

/// #8476: a route whose handler stalls answers `504` at the daemon's own
/// deadline instead of holding the connection until the caller gives up.
///
/// What: registers a control-plane session and holds its metadata write lock,
/// so `GET /api/v1/control/sessions` blocks on the read inside the handler.
/// Time is paused, so the deadline elapses at once. Before #8476 the request
/// never completed and the outer one-hour guard fired instead.
#[tokio::test(start_paused = true)]
async fn stalled_route_answers_504_at_the_server_deadline() {
    use crate::control::actor::SessionActorHandle;
    use crate::control::event::BackendKind;
    use crate::control::id::ControlSessionId;
    use crate::control::state::SessionMetadata;
    use axum::body::Body;
    use axum::extract::connect_info::MockConnectInfo;
    use axum::http::Request;
    use tower::ServiceExt;

    let (state, _dir) = hermetic_shared();
    let id = ControlSessionId::new("deadline-proj", 0);
    let (command_tx, _command_rx) = tokio::sync::mpsc::channel(4);
    let (event_tx, _) = tokio::sync::broadcast::channel(16);
    let metadata = Arc::new(tokio::sync::RwLock::new(SessionMetadata::new(
        id.clone(),
        "deadline-proj".into(),
        BackendKind::StreamJson,
    )));
    let handle = SessionActorHandle {
        command_tx,
        event_tx,
        write_lock_held: Arc::new(std::sync::atomic::AtomicBool::new(false)),
        metadata: Arc::clone(&metadata),
    };
    state.session_registry.register(id, handle).await;
    let _stall = metadata.write().await;

    let loopback = std::net::SocketAddr::from(([127, 0, 0, 1], 0));
    let app = router(Arc::clone(&state)).layer(MockConnectInfo(loopback));
    let started = tokio::time::Instant::now();
    let resp = tokio::time::timeout(
        std::time::Duration::from_secs(3600),
        app.oneshot(
            Request::builder()
                .uri("/api/v1/control/sessions")
                .body(Body::empty())
                .unwrap(),
        ),
    )
    .await
    .expect("the daemon must answer a stalled request itself, not leave it open for an hour")
    .unwrap();

    assert_eq!(resp.status(), StatusCode::GATEWAY_TIMEOUT);
    assert!(
        started.elapsed() < std::time::Duration::from_secs(30),
        "the deadline must fire before trusty-console's 30 s proxy bound; took {:?}",
        started.elapsed()
    );
    let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
        .await
        .unwrap();
    let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
    assert_eq!(body["route"], "/api/v1/control/sessions");
}

/// #8476: the deadline bounds the handler, never a streamed body — an SSE
/// stream still delivers an event long after the deadline has passed.
#[tokio::test(start_paused = true)]
async fn sse_stream_outlives_the_request_deadline() {
    use axum::body::Body;
    use axum::http::Request;
    use http_body_util::BodyExt;
    use tower::ServiceExt;

    let (state, id) = state_with_session();
    let request = Request::builder().uri("/events").body(Body::empty());
    let response = router(Arc::clone(&state))
        .oneshot(request.unwrap())
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::OK);

    // Well past every standard-class deadline.
    tokio::time::sleep(std::time::Duration::from_secs(120)).await;
    state
        .clone()
        .push_hook_event(crate::core::hook::HookEventRecord::now(
            id,
            HookEvent::PostToolUse,
            serde_json::json!({"tool": "Edit"}),
        ));

    let mut body = response.into_body();
    let text = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        loop {
            let frame = body.frame().await.expect("stream still open")?;
            if let Ok(data) = frame.into_data() {
                let text = String::from_utf8_lossy(&data).into_owned();
                // Keep-alive comments arrive too; wait for the event itself.
                if text.contains("data:") {
                    return Ok::<_, axum::Error>(text);
                }
            }
        }
    })
    .await
    .expect("the SSE event arrived after the deadline")
    .expect("frame read ok");
    assert!(text.contains("PostToolUse"), "unexpected frame: {text:?}");
}