trusty-mpm 1.8.2

//! Unit tests for [`super`]'s in-place-relaunch client (#2023 C, #2789).
//!
//! Why: extracted from an inline `#[cfg(test)] mod tests` so the production
//! `guided_inplace.rs` stays under the 500-SLOC cap while these tests keep the
//! 1500-SLOC test budget (basename `tests.rs`).
//! What: the local `TcpListener` HTTP mocks and every `plan_inplace_*` /
//! `reactivate_*` / `fetch_until_stopped_*` case, moved verbatim.
//! Test: this file IS the test module.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

use super::*;
use trusty_mpm::core::paths::FrameworkPaths;

const TEST_ID: &str = "11111111-2222-3333-4444-555555555555";

/// Spawn a background HTTP mock that replies `status_line` (+ empty JSON
/// body) to every connection, counting hits.
///
/// Why: `reactivate_managed_session`'s status-code handling (#2027) needs
/// a real HTTP round-trip to exercise reqwest's response parsing; this
/// mirrors `core::sm::providers::test_support`'s "read the full request
/// before replying" mock convention (that helper is `pub(crate)` to the
/// `trusty-mpm` LIB crate and unreachable from this `bin/tm` BINARY crate
/// — a separate compilation unit — hence the small inline copy in
/// [`read_full_request`] below, rather than pulling in an external
/// mock-server dependency for one test file).
/// What: binds an ephemeral port, loops accepting connections, and for
/// each one increments the returned counter, drains the request, then
/// replies. Runs until the test's tokio runtime shuts down.
/// Test: used by `reactivate_*` and the command-build-ordering test.
async fn spawn_mock(status_line: &'static str) -> (String, Arc<AtomicUsize>) {
    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
    let addr = listener.local_addr().expect("addr");
    let hits = Arc::new(AtomicUsize::new(0));
    let hits_task = hits.clone();
    tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            hits_task.fetch_add(1, Ordering::SeqCst);
            read_full_request(&mut sock).await;
            let body = "{}";
            let resp = format!(
                "{status_line}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body
            );
            let _ = sock.write_all(resp.as_bytes()).await;
            let _ = sock.shutdown().await;
        }
    });
    (format!("http://{addr}"), hits)
}

/// Read an entire HTTP/1.1 request (headers + any `Content-Length` body)
/// before the caller writes a response — avoids the connection-reset
/// flakiness a naive single `read` can cause (mirrors
/// `core::sm::providers::test_support::read_full_request`, unreachable
/// from this crate; see [`spawn_mock`]'s doc for why it is duplicated
/// here rather than shared).
async fn read_full_request(sock: &mut TcpStream) {
    let mut buf: Vec<u8> = Vec::with_capacity(4096);
    let mut chunk = [0u8; 4096];
    let header_end = loop {
        if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
            break pos + 4;
        }
        match sock.read(&mut chunk).await {
            Ok(0) => return,
            Ok(n) => buf.extend_from_slice(&chunk[..n]),
            Err(_) => return,
        }
    };
    let content_len: usize = String::from_utf8_lossy(&buf[..header_end])
        .split("\r\n")
        .find_map(|line| {
            let (name, value) = line.split_once(':')?;
            name.trim()
                .eq_ignore_ascii_case("content-length")
                .then(|| value.trim().parse().ok())?
        })
        .unwrap_or(0);
    let want_total = header_end + content_len;
    while buf.len() < want_total {
        match sock.read(&mut chunk).await {
            Ok(0) => return,
            Ok(n) => buf.extend_from_slice(&chunk[..n]),
            Err(_) => return,
        }
    }
}

/// Spawn a background HTTP mock that captures the FIRST request line (the
/// `METHOD /path?query HTTP/1.1` line) of the first connection into the
/// returned shared string, then replies `status_line`.
///
/// Why: [`spawn_mock`] drains the request but discards it, so it cannot
/// prove which URL was requested. The #2789 wire contract — the caller's
/// pane id reaching the daemon as a query param — is only meaningful if the
/// exact request target is observable; this helper captures it.
/// What: binds an ephemeral port; for the first connection reads until the
/// end of the request headers, records the request line, then replies with
/// `status_line` + an empty JSON body. Subsequent connections are still
/// answered but not re-captured.
/// Test: `reactivate_forwards_caller_pane_id_query`,
/// `reactivate_omits_caller_pane_id_query_when_absent`.
async fn spawn_capturing_mock(status_line: &'static str) -> (String, Arc<Mutex<String>>) {
    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
    let addr = listener.local_addr().expect("addr");
    let request_line = Arc::new(Mutex::new(String::new()));
    let request_line_task = request_line.clone();
    tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            let mut buf: Vec<u8> = Vec::with_capacity(4096);
            let mut chunk = [0u8; 4096];
            while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
                match sock.read(&mut chunk).await {
                    Ok(0) => break,
                    Ok(n) => buf.extend_from_slice(&chunk[..n]),
                    Err(_) => break,
                }
            }
            if let Ok(mut slot) = request_line_task.lock()
                && slot.is_empty()
            {
                *slot = String::from_utf8_lossy(&buf)
                    .lines()
                    .next()
                    .unwrap_or("")
                    .to_string();
            }
            let body = "{}";
            let resp = format!(
                "{status_line}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body
            );
            let _ = sock.write_all(resp.as_bytes()).await;
            let _ = sock.shutdown().await;
        }
    });
    (format!("http://{addr}"), request_line)
}

/// Spawn a background HTTP mock that replies 200 OK with `{"state": ..}`,
/// walking through `states` one entry per connection (clamping to the last
/// entry once exhausted) — used to simulate a record transitioning
/// `Active` -> `Stopped` across retries (#2148).
///
/// Why: [`fetch_managed_session_until_stopped`]'s retry behavior needs a
/// mock whose response changes across calls, unlike [`spawn_mock`]'s fixed
/// status line; this mirrors the same "read the full request before
/// replying" convention.
/// What: binds an ephemeral port, loops accepting connections, and for the
/// Nth connection replies with `states[min(N, states.len() - 1)]` as the
/// summary's `state` field. Runs until the test's tokio runtime shuts down.
/// Test: `fetch_until_stopped_*` below.
async fn spawn_state_mock(states: Vec<&'static str>) -> (String, Arc<AtomicUsize>) {
    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
    let addr = listener.local_addr().expect("addr");
    let hits = Arc::new(AtomicUsize::new(0));
    let hits_task = hits.clone();
    tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            let n = hits_task.fetch_add(1, Ordering::SeqCst);
            read_full_request(&mut sock).await;
            let idx = n.min(states.len().saturating_sub(1));
            let state = states.get(idx).copied().unwrap_or("active");
            let body = format!(r#"{{"id":"{TEST_ID}","name":"tm-test","state":"{state}"}}"#);
            let resp = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body
            );
            let _ = sock.write_all(resp.as_bytes()).await;
            let _ = sock.shutdown().await;
        }
    });
    (format!("http://{addr}"), hits)
}

/// The retry budget the `fetch_until_stopped_*` cases drive the loop with
/// (#5840).
///
/// Why: the production [`FETCH_RETRY_BUDGET`] is 400ms, and one loopback fetch
/// on a loaded CI runner can consume all of it — the loop then returns after a
/// single attempt, which is its documented contract but leaves "it retried"
/// untestable. A budget an order of magnitude above any plausible loopback
/// round trip makes the retry the code's decision rather than the scheduler's,
/// so the assertions below hold regardless of runner load.
/// What: 1 second, paired with the production 80ms
/// [`FETCH_RETRY_INTERVAL`] — roughly 12 polls, where the flake needed the
/// first fetch alone to blow the whole budget.
/// Test: the three `fetch_until_stopped_*` cases below.
const TEST_RETRY_BUDGET: std::time::Duration = std::time::Duration::from_secs(1);

/// A liveness ceiling for a call that must terminate on its own (#5840).
///
/// Why: "gives up rather than retrying forever" needs an upper bound, but a
/// TIGHT one (the former `budget + interval * 3`) fails on a slow runner
/// without a regression in sight — it measures the machine, not the code. A
/// bound 30x the budget still catches a loop that never exits, and cannot be
/// tripped by load.
const TEST_LIVENESS_CEILING: std::time::Duration =
    std::time::Duration::from_secs(TEST_RETRY_BUDGET.as_secs() * 30);

#[tokio::test]
async fn fetch_until_stopped_returns_immediately_when_already_stopped() {
    // #2148: the common case — no race — must not pay any retry delay.
    let (url, hits) = spawn_state_mock(vec!["stopped"]).await;
    let client = reqwest::Client::new();
    let record =
        fetch_managed_session_until_stopped(&client, &url, TEST_ID, TEST_RETRY_BUDGET).await;
    assert_eq!(record.map(|r| r.state), Some("stopped".to_string()));
    // #5840: one hit IS "paid no retry delay" — the loop's only delay is the
    // sleep between fetches, so a single fetch cannot have slept. The former
    // wall-clock `elapsed < budget` said the same thing while also failing
    // whenever the runner made one fetch slow.
    assert_eq!(
        hits.load(Ordering::SeqCst),
        1,
        "must not retry once the first fetch already reads stopped"
    );
}

#[tokio::test]
async fn fetch_until_stopped_retries_past_transitioning_state() {
    // #2148: the race this hardens against — the record briefly reads
    // "active" while the SessionEnd stop is still in flight, then settles
    // on "stopped" a couple of polls later.
    let (url, hits) = spawn_state_mock(vec!["active", "active", "stopped"]).await;
    let client = reqwest::Client::new();
    let record = tokio::time::timeout(
        TEST_LIVENESS_CEILING,
        fetch_managed_session_until_stopped(&client, &url, TEST_ID, TEST_RETRY_BUDGET),
    )
    .await
    .expect("must settle on stopped, not poll forever");
    assert_eq!(record.map(|r| r.state), Some("stopped".to_string()));
    assert!(
        hits.load(Ordering::SeqCst) >= 3,
        "must retry past the transitioning reads before accepting stopped"
    );
}

#[tokio::test]
async fn fetch_until_stopped_gives_up_after_budget_when_never_stopped() {
    // A genuinely non-stopped record (e.g. still active, or the id belongs
    // to some other running session) must not retry forever — it gives up
    // once the bounded budget elapses so the caller falls through promptly.
    let (url, hits) = spawn_state_mock(vec!["active"]).await;
    let client = reqwest::Client::new();
    let start = std::time::Instant::now();
    // #5840: the "does not retry forever" bound is a generous liveness
    // ceiling, not a tight window — a real regression (a loop with no exit)
    // trips it, a slow runner does not.
    let record = tokio::time::timeout(
        TEST_LIVENESS_CEILING,
        fetch_managed_session_until_stopped(&client, &url, TEST_ID, TEST_RETRY_BUDGET),
    )
    .await
    .expect("must give up once the budget elapses, not poll forever");
    let elapsed = start.elapsed();
    assert_eq!(record.map(|r| r.state), Some("active".to_string()));
    // A LOWER bound only gets safer under load, so this one stays wall-clock.
    assert!(
        elapsed >= TEST_RETRY_BUDGET,
        "must not give up before the retry budget elapses"
    );
    assert!(
        hits.load(Ordering::SeqCst) > 1,
        "must have retried at least once before giving up"
    );
}

#[test]
fn parse_show_environment_value_extracts_id() {
    assert_eq!(
        parse_show_environment_value(
            "TM_MANAGED_SESSION_ID=11111111-2222-3333-4444-555555555555\n"
        ),
        Some("11111111-2222-3333-4444-555555555555".to_string())
    );
}

#[test]
fn parse_show_environment_value_none_when_unset() {
    // tmux prints "-NAME" (no "=") when the variable is explicitly unset in
    // the session.
    assert_eq!(
        parse_show_environment_value("-TM_MANAGED_SESSION_ID\n"),
        None
    );
}

#[test]
fn parse_show_environment_value_none_when_empty() {
    assert_eq!(parse_show_environment_value(""), None);
    assert_eq!(
        parse_show_environment_value("TM_MANAGED_SESSION_ID=\n"),
        None
    );
}

// ── resolve_env_managed_session_id (#4061) ───────────────────────────────────
//
// Mutates the real process environment (`TM_MANAGED_SESSION_ID`, `TMUX`), so
// these are `#[serial_test::serial]` — mirroring the convention already used
// by env-mutating tests elsewhere in this binary (e.g.
// `tests_behavior_b_tests.rs`'s `REPOS_ROOT_ENV` tests) — and each restores
// whatever was there before via an RAII-style guard so no other test in the
// suite ever observes a leaked value.

/// Restores a named env var to its prior value (or removes it if it was
/// unset) when dropped — keeps env-mutating tests exception-safe and
/// leak-free regardless of assertion panics.
struct EnvVarGuard {
    key: &'static str,
    prev: Option<String>,
}

impl EnvVarGuard {
    fn set(key: &'static str, value: &str) -> Self {
        let prev = std::env::var(key).ok();
        // SAFETY: serialised by `#[serial_test::serial]` at every call site.
        unsafe { std::env::set_var(key, value) };
        Self { key, prev }
    }
}

impl Drop for EnvVarGuard {
    fn drop(&mut self) {
        // SAFETY: serialised by `#[serial_test::serial]` at every call site.
        unsafe {
            match &self.prev {
                Some(v) => std::env::set_var(self.key, v),
                None => std::env::remove_var(self.key),
            }
        }
    }
}

#[test]
#[serial_test::serial]
fn resolve_env_managed_session_id_prefers_process_env() {
    let _g1 = EnvVarGuard::set(MANAGED_SESSION_ID_ENV, TEST_ID);
    // Ensure the tmux fallback path is not what's answering — unset $TMUX so
    // a bug that skipped straight to the tmux branch could not accidentally
    // pass by returning None from a real `tmux` shell-out on this host.
    let _g2 = EnvVarGuard {
        key: "TMUX",
        prev: std::env::var("TMUX").ok(),
    };
    // SAFETY: serialised by `#[serial_test::serial]`.
    unsafe { std::env::remove_var("TMUX") };

    assert_eq!(
        resolve_env_managed_session_id(),
        Some(TEST_ID.to_string()),
        "the process env value must be returned even with no tmux session"
    );
}

#[test]
#[serial_test::serial]
fn resolve_env_managed_session_id_none_when_unset() {
    let _g1 = EnvVarGuard {
        key: MANAGED_SESSION_ID_ENV,
        prev: std::env::var(MANAGED_SESSION_ID_ENV).ok(),
    };
    // SAFETY: serialised by `#[serial_test::serial]`.
    unsafe { std::env::remove_var(MANAGED_SESSION_ID_ENV) };
    let _g2 = EnvVarGuard {
        key: "TMUX",
        prev: std::env::var("TMUX").ok(),
    };
    // SAFETY: serialised by `#[serial_test::serial]`.
    unsafe { std::env::remove_var("TMUX") };

    assert_eq!(
        resolve_env_managed_session_id(),
        None,
        "neither source has a value — must resolve to None, never a stale leak"
    );
}

#[test]
fn plan_inplace_selected_when_env_set_and_stopped() {
    assert_eq!(
        plan_inplace(Some(TEST_ID), Some("stopped"), true),
        Some(ResumeAction::InPlace)
    );
}

#[test]
fn plan_inplace_none_when_env_absent() {
    assert_eq!(plan_inplace(None, Some("stopped"), true), None);
    assert_eq!(plan_inplace(None, Some("active"), true), None);
    assert_eq!(plan_inplace(None, None, true), None);
}

#[test]
fn plan_inplace_none_when_env_set_but_unresolved() {
    // Guard case (#2023 C item 4): a stale/unknown id must fall through to
    // the ordinary guided picker rather than error.
    assert_eq!(plan_inplace(Some(TEST_ID), None, true), None);
}

#[test]
fn plan_inplace_selects_active_when_pane_confirmed() {
    // #9034 condition 2: an Active record whose recorded pane IS this pane is
    // the operator's own session — relaunch in place, not the picker.
    assert_eq!(
        plan_inplace(Some(TEST_ID), Some("active"), true),
        Some(ResumeAction::InPlace)
    );
}

#[test]
fn plan_inplace_none_when_resolved_but_not_stopped() {
    // Safety-boundary regression guard (#2027 code-critic WARN), narrowed by
    // #9034: an Active record only selects InPlace with a confirmed pane, so a
    // leaked/stale TM_MANAGED_SESSION_ID pointing at some OTHER running
    // session's pane still falls through to the ordinary guided picker.
    assert_eq!(
        plan_inplace(Some(TEST_ID), Some("active"), false),
        None,
        "an active record on another pane must not select InPlace"
    );
    // A stopped record on another pane falls through, as before #9034.
    assert_eq!(plan_inplace(Some(TEST_ID), Some("stopped"), false), None);
    // Every other state falls through whatever the pane says.
    for state in ["errored", "provisioning", "decommissioned"] {
        for pane_confirmed in [true, false] {
            assert_eq!(
                plan_inplace(Some(TEST_ID), Some(state), pane_confirmed),
                None,
                "state '{state}' (pane_confirmed={pane_confirmed}) must not select InPlace"
            );
        }
    }
}

/// Spawn a mock that answers every managed-session GET with one record in
/// `state`, bound to tmux pane `pane_id` (#9034).
///
/// Why: [`spawn_state_mock`]'s body carries no `pane_id`, and the pane gate
/// needs one to compare against.
/// What: replies 200 with `{id, name, state, pane_id}` to every connection,
/// counting hits. Runs until the test's tokio runtime shuts down.
/// Test: `try_inplace_active_record_other_pane_falls_through`.
async fn spawn_pane_record_mock(
    state: &'static str,
    pane_id: &'static str,
) -> (String, Arc<AtomicUsize>) {
    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
    let addr = listener.local_addr().expect("addr");
    let hits = Arc::new(AtomicUsize::new(0));
    let hits_task = hits.clone();
    tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            hits_task.fetch_add(1, Ordering::SeqCst);
            read_full_request(&mut sock).await;
            let body = format!(
                r#"{{"id":"{TEST_ID}","name":"tm-test","state":"{state}","pane_id":"{pane_id}"}}"#
            );
            let resp = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body
            );
            let _ = sock.write_all(resp.as_bytes()).await;
            let _ = sock.shutdown().await;
        }
    });
    (format!("http://{addr}"), hits)
}

#[tokio::test]
async fn try_inplace_active_record_other_pane_falls_through() {
    // #9034 condition 2: an Active record bound to pane %9, asked from pane %5,
    // is some other pane's live session — never this pane's to relaunch.
    let client = reqwest::Client::new();
    let (url, _hits) = spawn_pane_record_mock("active", "%9").await;
    let target = try_inplace_target(&client, &url, TEST_ID, || Some("%5".to_string())).await;
    assert!(
        target.is_none(),
        "an active record on another pane must fall through to the guided default"
    );
    // Control: the same record asked from its own pane IS selected, so the
    // fall-through above is the pane gate's verdict and not a fetch failure.
    let target = try_inplace_target(&client, &url, TEST_ID, || Some("%9".to_string()))
        .await
        .expect("an active record on this pane must relaunch in place");
    assert_eq!(target.record.state, "active");
    assert_eq!(target.current_pane_id.as_deref(), Some("%9"));
    assert!(
        target.pane_confirmed_dead,
        "an active record needs the proof-of-pane claim for the daemon's reconcile"
    );
}

#[tokio::test]
async fn reactivate_confirms_success_on_2xx() {
    let (url, hits) = spawn_mock("HTTP/1.1 200 OK").await;
    let client = reqwest::Client::new();
    let ok = reactivate_managed_session(&client, &url, TEST_ID, None, false).await;
    assert!(ok, "a 2xx response must confirm reactivation");
    assert_eq!(hits.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn reactivate_aborts_on_409_conflict() {
    // #2027 HIGH fix: a 409 (session not Stopped on the daemon's own
    // re-check) must NOT be treated as confirmed — the caller must abort
    // rather than proceeding to exec.
    let (url, _hits) = spawn_mock("HTTP/1.1 409 Conflict").await;
    let client = reqwest::Client::new();
    let ok = reactivate_managed_session(&client, &url, TEST_ID, None, false).await;
    assert!(!ok, "409 must NOT be treated as a confirmed reactivate");
}

#[tokio::test]
async fn reactivate_aborts_on_404_not_found() {
    let (url, _hits) = spawn_mock("HTTP/1.1 404 Not Found").await;
    let client = reqwest::Client::new();
    let ok = reactivate_managed_session(&client, &url, TEST_ID, None, false).await;
    assert!(!ok, "404 must NOT be treated as a confirmed reactivate");
}

#[tokio::test]
async fn reactivate_aborts_on_unreachable_daemon() {
    // Nothing listens on a privileged low port -> immediate connection
    // refused, exercising the network-error branch without waiting out
    // the full PROBE_TIMEOUT.
    let client = reqwest::Client::new();
    let ok = reactivate_managed_session(&client, "http://127.0.0.1:1", TEST_ID, None, false).await;
    assert!(
        !ok,
        "an unreachable daemon must NOT be treated as a confirmed reactivate"
    );
}

#[tokio::test]
async fn reactivate_forwards_caller_pane_id_query() {
    // #2789: the caller's own tmux pane_id must reach the daemon as a
    // `?caller_pane_id=` query param — that is the whole mechanism by which
    // the daemon learns to treat the requesting `tm` pane as idle instead
    // of 409'ing. Capture the raw request line and assert the (URL-encoded)
    // pane id is present.
    let (url, request_line) = spawn_capturing_mock("HTTP/1.1 200 OK").await;
    let client = reqwest::Client::new();
    let ok = reactivate_managed_session(&client, &url, TEST_ID, Some("%3"), false).await;
    assert!(ok, "a 2xx response must confirm reactivation");
    let line = request_line.lock().expect("lock").clone();
    assert!(
        line.contains("caller_pane_id=%253"),
        "the request line must carry the URL-encoded caller_pane_id (%3 -> %253); got: {line}"
    );
    assert!(
        !line.contains("pane_confirmed_dead"),
        "pane_confirmed_dead=false must not be forwarded; got: {line}"
    );
}

#[tokio::test]
async fn reactivate_forwards_pane_confirmed_dead_query() {
    // #2794: when the caller has confirmed pane identity (proof-of-death), the
    // request must carry `?pane_confirmed_dead=true` so the daemon's reconcile
    // overrides its tmux-pane liveness probe for the caller's OWN pane.
    let (url, request_line) = spawn_capturing_mock("HTTP/1.1 200 OK").await;
    let client = reqwest::Client::new();
    let ok = reactivate_managed_session(&client, &url, TEST_ID, Some("%3"), true).await;
    assert!(ok, "a 2xx response must confirm reactivation");
    let line = request_line.lock().expect("lock").clone();
    assert!(
        line.contains("pane_confirmed_dead=true"),
        "a pane-confirmed-dead caller must forward the proof-of-death param; got: {line}"
    );
    assert!(
        line.contains("caller_pane_id=%253"),
        "the caller_pane_id must still accompany the proof-of-death param; got: {line}"
    );
}

#[tokio::test]
async fn reactivate_omits_caller_pane_id_query_when_absent() {
    // Backward compatibility: a non-tmux / pre-#2789 caller sends no
    // caller_pane_id, and the request must carry no such query param.
    let (url, request_line) = spawn_capturing_mock("HTTP/1.1 200 OK").await;
    let client = reqwest::Client::new();
    let ok = reactivate_managed_session(&client, &url, TEST_ID, None, false).await;
    assert!(ok);
    let line = request_line.lock().expect("lock").clone();
    assert!(
        !line.contains("caller_pane_id"),
        "no caller_pane_id query param may be sent when the caller passes None; got: {line}"
    );
}

/// A `ManagedSessionSummary` whose workspace is `workspace`, in the `stopped`
/// shape `run_inplace_relaunch` is normally handed.
///
/// Why: the struct has ~20 fields and three #4204 tests need it; inlining it
/// three more times would be pure noise.
/// What: `TEST_ID`/`tm-test`/`stopped` with `workspace_path` set and every
/// other field at its empty default.
/// Test: used by `run_inplace_relaunch_falls_through_on_gutted_worktree` and
/// `run_inplace_relaunch_serves_live_linked_worktree`.
fn stopped_record_at(workspace: &std::path::Path) -> trusty_mpm::client::ManagedSessionSummary {
    trusty_mpm::client::ManagedSessionSummary {
        id: TEST_ID.to_string(),
        name: "tm-test".to_string(),
        state: "stopped".to_string(),
        persisted_state: None,
        workspace_path: Some(workspace.to_string_lossy().to_string()),
        repo_url: None,
        branch: None,
        created_at: None,
        last_activity_at: None,
        pending_decision: None,
        proposed_default: None,
        source_id: None,
        task: None,
        cwd: None,
        claude_session_id: None,
        deliverable_id: None,
        pane_id: None,
        injection_status: None,
        unresumable: false,
        stale_assets: false,
        stale_assets_unchecked: false,
        attached: false,
        slot: 0,
        deleted: false,
        auto_resume_parked: None,
        kind: None,
    }
}

#[tokio::test]
async fn run_inplace_relaunch_falls_through_on_gutted_worktree() {
    // ISSUE #4204: a worktree whose `.git` and source tree were stripped keeps
    // its directory node, so the old `cwd.is_dir()` gate passed and this
    // function exec'd `claude` into the husk (observed 2026-07-27: PID 63302
    // plus four MCP children parked on the dead cwd of
    // `.base/.worktrees/f443c12d-…`). It must now fall through to the picker
    // instead, BEFORE any daemon mutation.
    //
    // The mock answers 409 so that even on pre-fix code — where `claude` may
    // resolve and the flow would continue — the reactivate is REFUSED and the
    // real `exec` (which would replace this test process) is never reached.
    // The decisive assertion is therefore `hits == 0`, not the outcome alone:
    // against main this test fails either way — with `claude` absent the
    // outcome is `Result(Err(..))`, and with `claude` present the daemon is
    // contacted and `hits == 1`.
    let base = tempfile::tempdir().expect("tempdir");
    let gutted = base
        .path()
        .join(".worktrees")
        .join("f443c12d-2fb6-4ce1-9f70-2e7695306e47");
    std::fs::create_dir_all(gutted.join(".claude")).expect("create gutted worktree");
    assert!(
        gutted.is_dir() && !gutted.join(".git").exists(),
        "fixture must be exactly what the old is_dir() gate waved through"
    );

    let (url, hits) = spawn_mock("HTTP/1.1 409 Conflict").await;
    let client = reqwest::Client::new();

    let fw_home = crate::test_support::hermetic_temp_dir();
    let outcome = run_inplace_relaunch(
        &client,
        &url,
        TEST_ID,
        stopped_record_at(&gutted),
        None,
        false,
        &FrameworkPaths::under(fw_home.path()),
    )
    .await;

    assert!(
        matches!(outcome, InPlaceOutcome::FallThrough),
        "a gutted worktree must fall through to the picker, never be exec'd into"
    );
    assert_eq!(
        hits.load(Ordering::SeqCst),
        0,
        "the liveness gate must abort BEFORE any daemon mutation (#2790/#4204)"
    );
}

#[tokio::test]
async fn run_inplace_relaunch_serves_live_linked_worktree() {
    // THE ANTI-OVER-REFUSAL COUNTERPART. In a linked worktree `.git` is a FILE
    // holding a `gitdir:` pointer — every managed workspace looks like this. A
    // guard that assumed a `.git` DIRECTORY would fall through for all of them,
    // silently breaking in-place relaunch entirely.
    //
    // The observable proof that the gate was PASSED is that control reached the
    // next step: either command resolution failed (no `claude` on this machine)
    // or the daemon was contacted. The one outcome that must NOT occur is the
    // gutted signature — `FallThrough` with zero daemon hits.
    let base = tempfile::tempdir().expect("tempdir");
    let live = base
        .path()
        .join(".worktrees")
        .join("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee");
    std::fs::create_dir_all(&live).expect("create worktree");
    std::fs::write(
        live.join(".git"),
        "gitdir: /repo/.base/.git/worktrees/live\n",
    )
    .expect("write .git pointer file");
    assert!(
        live.join(".git").is_file() && !live.join(".git").is_dir(),
        "fixture must model the linked-worktree .git FILE this test exists for"
    );

    let (url, hits) = spawn_mock("HTTP/1.1 409 Conflict").await;
    let client = reqwest::Client::new();

    // #8545: provision under a temp home, never the operator's.
    let fw_home = crate::test_support::hermetic_temp_dir();
    let outcome = run_inplace_relaunch(
        &client,
        &url,
        TEST_ID,
        stopped_record_at(&live),
        None,
        false,
        &FrameworkPaths::under(fw_home.path()),
    )
    .await;

    let build_failed = matches!(outcome, InPlaceOutcome::Result(Err(_)));
    let daemon_contacted = hits.load(Ordering::SeqCst) > 0;
    assert!(
        build_failed || daemon_contacted,
        "a LIVE linked worktree must get past the liveness gate — it fell through \
         with the daemon never contacted, which is the gutted-workspace verdict"
    );
}

#[tokio::test]
async fn run_inplace_relaunch_never_reactivates_when_command_build_fails() {
    // #2027 MEDIUM fix: command resolution (resolve `claude` + build the
    // resume argv) must happen BEFORE the daemon is ever asked to
    // reactivate the record, so a missing-binary failure never flips a
    // Stopped record to a false Active.
    //
    // This ordering guarantee can only be exercised end-to-end on a
    // machine where `claude` is NOT resolvable — otherwise
    // build_inplace_resume_command succeeds and this test would be
    // vacuously true. Probe first; skip (don't fail) when claude IS
    // present, mirroring the inverse of the "skip when claude absent"
    // convention used throughout runtime::claude_code's own test suite.
    let tmp = tempfile::tempdir().expect("tempdir");
    // #8545: provision under a temp home, never the operator's.
    let fw_home = crate::test_support::hermetic_temp_dir();
    let fw = FrameworkPaths::under(fw_home.path());
    if trusty_mpm::runtime::build_inplace_resume_command_under(&fw, tmp.path(), None).is_ok() {
        return;
    }

    let (url, hits) = spawn_mock("HTTP/1.1 200 OK").await;
    let record = stopped_record_at(tmp.path());
    let client = reqwest::Client::new();

    let outcome = run_inplace_relaunch(&client, &url, TEST_ID, record, None, false, &fw).await;

    assert!(
        matches!(outcome, InPlaceOutcome::Result(Err(_))),
        "a command-build failure must surface as a hard error, not FallThrough"
    );
    assert_eq!(
        hits.load(Ordering::SeqCst),
        0,
        "reactivate must NEVER be called when command resolution fails first (#2027)"
    );
}

// ── #4336: the exec seam's argv/environment ───────────────────────────────

/// Build a synthetic [`trusty_mpm::runtime::InPlaceResumeCommand`] so the
/// exec-seam assertions below run on every machine.
///
/// Why: `build_inplace_resume_command` needs a real `claude` install, so a
/// test built only on top of it is skipped on CI — exactly the coverage hole
/// #4336 was reported into. Hand-constructing the struct pins
/// [`build_inplace_exec_command`]'s own contract (does it forward EVERY arg,
/// the cwd, and the env?) unconditionally.
/// What: a fixed binary path plus `args`, with a config dir and oauth token.
/// Test: used by the `inplace_exec_command_*` cases below.
fn synthetic_resume(args: &[&str]) -> trusty_mpm::runtime::InPlaceResumeCommand {
    trusty_mpm::runtime::InPlaceResumeCommand {
        mcp_env: Vec::new(),
        claude_bin: "/fake/bin/claude".to_owned(),
        args: args.iter().map(|s| (*s).to_owned()).collect(),
        config_dir: Some(std::path::PathBuf::from("/fake/config")),
        oauth_token: Some("sk-ant-oat01-fake".to_owned()),
        profile: Default::default(),
    }
}

/// The unbound `gh` identity — an operator with no `github:` binding.
///
/// Why: most `inplace_exec_command_*` cases are about argv and auth env, not
/// about `gh`, and an empty [`GhEnv`] is exactly what they saw before #8233
/// gave this seam a `gh_env` parameter.
/// Test: used by every `inplace_exec_command_*` case that is not about `gh`.
fn no_gh() -> trusty_mpm::core::gh_identity::GhEnv {
    trusty_mpm::core::gh_identity::GhEnv::default()
}

/// A `gh` identity bound by `config_dir`, resolved through the production
/// precedence engine rather than hand-built.
///
/// Why: asserting against a hand-assembled `GhEnv` would prove only that this
/// function copies a struct. Resolving a real [`GithubConfig`] means the test
/// breaks if the precedence or the #6668 removal set changes.
/// Test: `inplace_exec_command_carries_the_pinned_gh_identity`,
/// `inplace_exec_command_clears_an_inherited_gh_token`.
fn bound_gh(config_dir: &str) -> trusty_mpm::core::gh_identity::GhEnv {
    let cfg = trusty_mpm::core::trusty_tools_config::GithubConfig {
        config_dir: Some(config_dir.into()),
        token_env: None,
        account: None,
        host: None,
    };
    trusty_mpm::core::gh_identity::resolve_gh_env(Some(&cfg))
        .expect("a config_dir binding resolves")
}

/// Every `(name, value)` the built command will apply, with `None` for a removal.
fn env_of(cmd: &std::process::Command) -> Vec<(String, Option<String>)> {
    cmd.get_envs()
        .map(|(k, v)| {
            (
                k.to_string_lossy().into_owned(),
                v.map(|v| v.to_string_lossy().into_owned()),
            )
        })
        .collect()
}

/// #8233 review (HIGH): the bare-`tm` in-place relaunch used to inherit
/// `GH_CONFIG_DIR`/`GH_TOKEN` from the pane shell, which the pre-#8233 spawn had
/// exported there by sourcing a temp file. The launch spec delivers the
/// environment to `claude` alone now, so this seam must resolve and apply the
/// binding itself or every in-place relaunch silently runs as the ambient
/// account.
#[test]
fn inplace_exec_command_carries_the_pinned_gh_identity() {
    let resume = synthetic_resume(&["--dangerously-skip-permissions"]);
    let cmd = build_inplace_exec_command(
        &resume,
        std::path::Path::new("/fake/cwd"),
        &bound_gh("/cfg/pinned"),
        false,
    );

    let envs = env_of(&cmd);
    assert!(
        envs.contains(&("GH_CONFIG_DIR".to_owned(), Some("/cfg/pinned".to_owned()))),
        "the relaunched claude must see the project's gh config home: {envs:?}"
    );
}

/// #6668, for the path #8233 reintroduced it on: `gh` reads an env token BEFORE
/// a config dir, so applying the binding without also removing an inherited
/// `GH_TOKEN` leaves the wrong account in force for the session's whole life.
#[test]
fn inplace_exec_command_clears_an_inherited_gh_token() {
    let resume = synthetic_resume(&["--dangerously-skip-permissions"]);
    let cmd = build_inplace_exec_command(
        &resume,
        std::path::Path::new("/fake/cwd"),
        &bound_gh("/cfg/pinned"),
        false,
    );

    let envs = env_of(&cmd);
    assert!(
        envs.contains(&("GH_TOKEN".to_owned(), None)),
        "an inherited GH_TOKEN must be REMOVED, not merely outranked: {envs:?}"
    );
}

#[test]
fn inplace_exec_command_forwards_every_arg_in_order() {
    // #4336 core regression: the Command handed to `exec` must carry the
    // composed argv VERBATIM. An empty (or truncated) `get_args()` here is
    // precisely the reported "execs claude with zero args" defect.
    // #6765: `--resume <id>`, not `--continue` — the composer no longer emits a
    // bare `--continue`, so the sample argv tracks what it really produces.
    let resume = synthetic_resume(&[
        "--setting-sources",
        "project,local",
        "--dangerously-skip-permissions",
        "--resume",
        "abc-123",
    ]);
    let cmd =
        build_inplace_exec_command(&resume, std::path::Path::new("/fake/cwd"), &no_gh(), false);

    let args: Vec<String> = cmd
        .get_args()
        .map(|a| a.to_string_lossy().into_owned())
        .collect();
    assert_eq!(
        args,
        vec![
            "--setting-sources",
            "project,local",
            "--dangerously-skip-permissions",
            "--resume",
            "abc-123"
        ],
        "the exec seam must forward the composed argv verbatim and in order"
    );
    assert_eq!(
        cmd.get_program().to_string_lossy(),
        "/fake/bin/claude",
        "the resolved claude binary must be the exec target"
    );
    assert_eq!(
        cmd.get_current_dir(),
        Some(std::path::Path::new("/fake/cwd")),
        "the relaunch must be rooted at the record's workspace"
    );
}

/// #4181 / ADR-0042: a non-empty `mcp_env` reaches the exec'd `Command`.
///
/// Why: this path replaces the process with `claude` directly, so the `env
/// NAME=VALUE` prefix the pane paths build never applies — the pins must be set
/// on the `Command` itself or an in-place relaunch silently loses them while the
/// other two spawn paths keep them. `synthetic_resume` leaves `mcp_env` empty,
/// so nothing here covered the carrier.
/// Test: itself.
#[test]
fn inplace_exec_command_carries_a_non_empty_mcp_env() {
    let mut resume = synthetic_resume(&["--dangerously-skip-permissions"]);
    resume.mcp_env = vec![
        (
            "TRUSTY_MEMORY_PALACE".to_owned(),
            "owner repo slug".to_owned(),
        ),
        ("TRUSTY_INDEX".to_owned(), "idx-42".to_owned()),
    ];
    let cmd =
        build_inplace_exec_command(&resume, std::path::Path::new("/fake/cwd"), &no_gh(), false);

    let envs: Vec<(String, Option<String>)> = cmd
        .get_envs()
        .map(|(k, v)| {
            (
                k.to_string_lossy().into_owned(),
                v.map(|v| v.to_string_lossy().into_owned()),
            )
        })
        .collect();

    // No shell parses this, so the value must arrive VERBATIM — spaces and all,
    // with no quoting added.
    assert!(
        envs.contains(&(
            "TRUSTY_MEMORY_PALACE".to_owned(),
            Some("owner repo slug".to_owned())
        )),
        "the palace pin must be set on the exec'd Command verbatim: {envs:?}"
    );
    assert!(
        envs.contains(&("TRUSTY_INDEX".to_owned(), Some("idx-42".to_owned()))),
        "the index pin must be set on the exec'd Command: {envs:?}"
    );
    // The #4455 guard still holds alongside them.
    assert!(
        envs.contains(&(
            "CLAUDE_CONFIG_DIR".to_owned(),
            Some("/fake/config".to_owned())
        )),
        "the pins must not displace CLAUDE_CONFIG_DIR: {envs:?}"
    );
}

#[test]
fn inplace_exec_command_scrubs_api_key_and_sets_auth_env() {
    // The env invariants `env_bin_prefix` encodes for the shell-string paths
    // must hold identically on the exec path (DOC-34 + #2246).
    let resume = synthetic_resume(&["--dangerously-skip-permissions"]);
    let cmd =
        build_inplace_exec_command(&resume, std::path::Path::new("/fake/cwd"), &no_gh(), false);

    let envs: Vec<(String, Option<String>)> = cmd
        .get_envs()
        .map(|(k, v)| {
            (
                k.to_string_lossy().into_owned(),
                v.map(|v| v.to_string_lossy().into_owned()),
            )
        })
        .collect();

    assert!(
        envs.contains(&("ANTHROPIC_API_KEY".to_owned(), None)),
        "ANTHROPIC_API_KEY must be REMOVED (None) for the relaunched claude: {envs:?}"
    );
    assert!(
        envs.contains(&(
            "CLAUDE_CONFIG_DIR".to_owned(),
            Some("/fake/config".to_owned())
        )),
        "the tm-owned CLAUDE_CONFIG_DIR must be injected: {envs:?}"
    );
    assert!(
        envs.iter().any(
            |(k, v)| k == trusty_mpm::core::oauth_token::OAUTH_TOKEN_ENV_VAR
                && v.as_deref() == Some("sk-ant-oat01-fake")
        ),
        "#2246: the resolved oauth token must be injected: {envs:?}"
    );
}

#[test]
fn inplace_exec_command_scrubs_inherited_session_markers() {
    // #4467: a bare `tm` relaunch run from inside a Claude Code session inherits
    // CLAUDE_CODE_CHILD_SESSION and, without this scrub, the relaunched session
    // saves no transcript — no --resume, no --continue, no /rewind. The marker
    // name is hard-coded so this cannot pass vacuously if the shared marker list
    // is emptied.
    let resume = synthetic_resume(&["--dangerously-skip-permissions"]);
    let cmd =
        build_inplace_exec_command(&resume, std::path::Path::new("/fake/cwd"), &no_gh(), false);

    let envs: Vec<(String, Option<String>)> = cmd
        .get_envs()
        .map(|(k, v)| {
            (
                k.to_string_lossy().into_owned(),
                v.map(|v| v.to_string_lossy().into_owned()),
            )
        })
        .collect();

    assert!(
        envs.contains(&("CLAUDE_CODE_CHILD_SESSION".to_owned(), None)),
        "the transcript-suppressing marker must be REMOVED (None): {envs:?}"
    );
    assert!(
        envs.contains(&("CLAUDE_CODE_SESSION_ID".to_owned(), None)),
        "the parent's session id must be REMOVED (None): {envs:?}"
    );
    // Over-scrub guard: the scrub runs BEFORE the deliberate assignments, so
    // CLAUDE_CONFIG_DIR must survive it as a SET value (#4455 / #4451).
    assert!(
        envs.contains(&(
            "CLAUDE_CONFIG_DIR".to_owned(),
            Some("/fake/config".to_owned())
        )),
        "CLAUDE_CONFIG_DIR must survive the scrub as a set value: {envs:?}"
    );
}

/// #8405: the in-place relaunch assigns the renderer config decides, in both
/// directions, whatever this pane exports. At df212601d the builder yielded to
/// an exported value, so this fails there under any ambient environment: one
/// of the two directions disagrees with it.
#[test]
fn inplace_exec_command_assigns_the_configured_renderer() {
    use trusty_mpm::core::alt_screen::ALT_SCREEN_ENV_VAR;

    let resume = synthetic_resume(&["--dangerously-skip-permissions"]);
    for (alternate_screen, want) in [(true, "0"), (false, "1")] {
        let cmd = build_inplace_exec_command(
            &resume,
            std::path::Path::new("/fake/cwd"),
            &no_gh(),
            alternate_screen,
        );
        let carried = cmd
            .get_envs()
            .find(|(k, _)| *k == ALT_SCREEN_ENV_VAR)
            .and_then(|(_, v)| v.map(|v| v.to_string_lossy().into_owned()));
        assert_eq!(
            carried.as_deref(),
            Some(want),
            "alternate_screen={alternate_screen}"
        );
    }
}

/// #7160: the mouse-capture default still yields to an exported value.
#[test]
fn inplace_exec_command_defaults_the_mouse_capture_off() {
    use trusty_mpm::core::alt_screen::{MOUSE_DEFAULT, MOUSE_ENV_VAR};

    let resume = synthetic_resume(&["--dangerously-skip-permissions"]);
    let cmd =
        build_inplace_exec_command(&resume, std::path::Path::new("/fake/cwd"), &no_gh(), false);

    let carried_by_the_launch = std::env::var_os(MOUSE_ENV_VAR).is_some();
    let provisioned = cmd.get_envs().any(|(k, v)| {
        k == MOUSE_ENV_VAR && v.is_some_and(|v| v == std::ffi::OsStr::new(MOUSE_DEFAULT))
    });
    assert_eq!(
        provisioned, !carried_by_the_launch,
        "tm must provision {MOUSE_ENV_VAR}={MOUSE_DEFAULT} when the launch \
         carries no value, and leave an operator value untouched when it does"
    );
}

#[serial_test::serial]
#[test]
fn inplace_exec_command_carries_isolation_flags_and_persona_end_to_end() {
    // #4336 end-to-end: composed through the REAL builder (not a synthetic
    // struct), the argv that reaches `exec` must carry both isolation flags
    // AND the PM system prompt. Before this fix `--append-system-prompt-file`
    // was omitted by design, so an in-place relaunch restored the operator
    // into vanilla Claude Code. Requires a real `claude` install; skip
    // otherwise, matching this file's established convention.
    let tmp = tempfile::tempdir().expect("tempdir");
    // #8545: provision under a temp home, never the operator's.
    let fw_home = crate::test_support::hermetic_temp_dir();
    let fw = FrameworkPaths::under(fw_home.path());
    let Ok(resume) = trusty_mpm::runtime::build_inplace_resume_command_under(&fw, tmp.path(), None)
    else {
        return;
    };
    let cmd = build_inplace_exec_command(&resume, tmp.path(), &no_gh(), false);
    let args: Vec<String> = cmd
        .get_args()
        .map(|a| a.to_string_lossy().into_owned())
        .collect();

    // #4451: `build_inplace_resume_command` always resolves a tm-owned
    // CLAUDE_CONFIG_DIR, so the relocated tier list is the correct expectation
    // here — `user` names tm's own config home, not the operator's ~/.claude,
    // so the #1269 isolation this test guards is unchanged.
    assert!(
        args.windows(2)
            .any(|w| w == ["--setting-sources", "user,project,local"]),
        "--setting-sources user,project,local must survive to exec (#1269/#4451): {args:?}"
    );
    assert!(
        args.iter().any(|a| a == "--dangerously-skip-permissions"),
        "--dangerously-skip-permissions must survive to exec (#1269): {args:?}"
    );
    let prompt_idx = args
        .iter()
        .position(|a| a == "--append-system-prompt-file")
        .expect("in-place relaunch must carry the PM system prompt (#4336)");
    let prompt_path = args
        .get(prompt_idx + 1)
        .expect("--append-system-prompt-file must be followed by a path");
    assert!(
        std::path::Path::new(prompt_path).is_file(),
        "the prompt-file argv token must name a readable file, unquoted: {prompt_path}"
    );
}

/// #8405: the relaunch seam reads the renderer from its config root, both
/// directions. Fails if the seam ignores the config.
#[test]
fn inplace_exec_command_for_follows_the_configured_renderer() {
    use trusty_mpm::core::alt_screen::ALT_SCREEN_ENV_VAR;

    let resume = synthetic_resume(&["--dangerously-skip-permissions"]);
    for (alternate_screen, want) in [(true, "0"), (false, "1")] {
        let root = crate::test_support::config_root_with_alternate_screen(alternate_screen);
        let cmd = inplace_exec_command_for(
            Some(root.path()),
            &resume,
            std::path::Path::new("/fake/cwd"),
            &no_gh(),
        );
        let carried = cmd
            .get_envs()
            .find(|(k, _)| *k == ALT_SCREEN_ENV_VAR)
            .and_then(|(_, v)| v.map(|v| v.to_string_lossy().into_owned()));
        assert_eq!(
            carried.as_deref(),
            Some(want),
            "alternate_screen={alternate_screen}"
        );
    }
}