//! RC-RENDER-TRUTH W1 — the KNOWN-HAZARDS 7.46 umbrella regression matrix
//! (ADR-0042): an rc surface answers from LIVE session authority, never a stale
//! persisted projection; a resuming perch is UNBOUND, not offline. Driven
//! against the REAL `spt` binary + a REAL `spt daemon run` tree + the
//! dummy-harness fixture — no mocks (the v0.12.1 real-harness ruling). Shape
//! source: hertz's organic pre-edit perri row (`status=offline` over a live
//! session, `perri-info.pre-unbound-manual.json`).
//!
//! Matrix rows here:
//!   1. offline row + honest live session  ⇒ `spt rc` ATTACHES (reader truth).
//!   2. offline row + no session           ⇒ the existing offline refusal.
//!   3. claimed session, dead client tree  ⇒ refusal, NEVER an attach.
//!   4. resume-never-bound                 ⇒ perch reads UNBOUND (writer truth)
//!      and the live session is attachable.
//!   5. harness-only endpoint (local row + gossiped remote Instance) ⇒ truthful
//!      pre-stream refusal naming the harness-hosted state, never a stale-row
//!      guess.
//!
//! The qualified-target row (bare + `id@node` + `subnet:id` against a REMOTE
//! broker-hosted target, wire always bare) rides the in-crate two-broker rig in
//! `rc::tests` — establish_attach is crate-private.
//!
//! HEAVY nextest group at birth (real daemon trees): ci.yml Phase A/Phase B
//! and nextest.toml, three identical strings. Bounded; every spawned pid
//! reaped scoped (never machine-wide — shared runner).

use std::io::Read;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Output, Stdio};
use std::sync::Mutex;
use std::time::{Duration, Instant};

mod common;
use common::CommandNoWindowExt;

use spt_store::perch::{self, ParentHint};

// One file = one test binary: the tests set the PROCESS-GLOBAL SPT_HOME then do
// in-proc reads (perch paths, registry::register, seed_subnets) — serialize each
// whole body on one file-local lock (the multi_subnet_bringup_e2e.rs pattern).
static E2E_LOCK: Mutex<()> = Mutex::new(());

fn kill_pid(pid: u32) {
    #[cfg(windows)]
    let _ = Command::new("taskkill")
        .no_window()
        .args(["/PID", &pid.to_string(), "/F", "/T"])
        .output();
    #[cfg(unix)]
    let _ = Command::new("kill").args(["-9", &pid.to_string()]).output();
}



fn ready_pid(path: &Path) -> Option<u32> {
    let s = std::fs::read_to_string(path).ok()?;
    serde_json::from_str::<serde_json::Value>(&s)
        .ok()?
        .get("pid")?
        .as_u64()
        .map(|p| p as u32)
}

fn wait_for_ready_pid(path: &Path, budget: Duration) -> Option<u32> {
    let deadline = Instant::now() + budget;
    while Instant::now() < deadline {
        if let Some(p) = ready_pid(path) {
            return Some(p);
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    None
}



/// Seed `names` as subnets this node is a member of (single name → auto-home,
/// keeping the focus on the attach truth under test).
fn seed_subnets(names: &[&str]) {
    let mut store = spt_store::subnet::SubnetStore::load();
    for n in names {
        store.create_subnet(n, spt_store::access::Mode::Open).expect("seed subnet");
    }
    store.save().expect("save subnets");
}

/// Register a harness adapter `name` whose `[session.self]` runs the dummy in
/// `mode` (`dummy` binds → ONLINE; `hold-unbound` holds a LIVE session that
/// never binds). Mirrors multi_subnet_bringup_e2e's fixture registration.
fn register_harness(home: &Path, spt_bin: &Path, mock: &Path, name: &str, mode: &str) {
    let exe_suffix = spt_bin
        .extension()
        .map(|e| format!(".{}", e.to_string_lossy()))
        .unwrap_or_default();
    let spt_for_cmd = spt_bin.to_string_lossy().replace('\\', "/");
    let mock_for_cmd = mock.to_string_lossy().replace('\\', "/");
    let self_cmd = format!(
        "{mock_for_cmd} --mode {mode} --id {{id}} --session-id {{session_id}} \
         --spt {spt_for_cmd} --mock-adapter {name} --type live_agent --heartbeat-ms 200"
    );
    register_raw_harness(home, spt_bin, name, &self_cmd, exe_suffix);
}

/// Register a harness adapter with a RAW `[session.self]` command — the zombie
/// rig's entry (a wrapper command that survives its client's death).
fn register_raw_harness(home: &Path, spt_bin: &Path, name: &str, self_cmd: &str, exe_suffix: String) {
    let src = perch::spt_home().join("srcs").join(name);
    std::fs::create_dir_all(&src).unwrap();
    let psyche_bin = src.join(format!("psychebin{exe_suffix}"));
    std::fs::copy(spt_bin, &psyche_bin).expect("stage the psyche binary fixture");
    let drops = home.join("drops");
    std::fs::create_dir_all(&drops).unwrap();
    let dir = drops.to_string_lossy().replace('\\', "/");
    let manifest_toml = format!(
        "[adapter]\nname=\"{name}\"\nkind=\"harness\"\nversion=\"1\"\n\
         min_spt_core_version=\"0\"\n\n[session.self]\ncommand='{self_cmd}'\n\n\
         [session]\ncommune_dir=\"{dir}\"\nsignoff_dir=\"{dir}\"\n\n\
         [session.psyche_init]\ncommand='psychebin ready {{id}}'\ncwd=\"{{psyche_dir}}\"\nkeys=[]\n"
    );
    std::fs::write(src.join("manifest.toml"), &manifest_toml).unwrap();
    spt_runtime::registry::register(&perch::adapters_dir(), &src, 1000).unwrap();
}

/// Spawn the REAL daemon broker; return `(child, brain_pid, brain_log_path)`.
fn spawn_broker(home: &Path, spt_bin: &Path) -> (Child, u32, PathBuf) {
    let brain_log = home.join("brain.stderr.log");
    let brain_log_file = std::fs::File::create(&brain_log).expect("create brain stderr log");
    let mut broker: Child = Command::new(spt_bin)
        .no_window()
        .args(["daemon", "run"])
        .env("SPT_HOME", home)
        // Rig daemon: the advisory well-known ports belong to this box's resident
        // fleet, never to a test tree (REQ-TEST-DAEMON-EPHEMERAL-ADVISORY-PORTS).
        .env(
            spt_daemon::docshost::TEST_EPHEMERAL_ADVISORY_PORTS_ENV,
            "1",
        )
        .stdout(Stdio::null())
        .stderr(Stdio::from(brain_log_file))
        .spawn()
        .expect("spawn spt daemon run (broker process)");
    match wait_for_ready_pid(&home.join("brain.ready"), Duration::from_secs(30)) {
        Some(p) => (broker, p, brain_log),
        None => {
            let _ = broker.kill();
            let _ = broker.wait();
            panic!(
                "PRECONDITION: brain never came up.\n{}",
                common::daemon_stderr_panel(&brain_log)
            );
        }
    }
}

/// Scoped reap: daemon-stop, then kill the broker + any extra pids + a hosted
/// `{id}-psyche` for EVERY endpoint the test started. Never machine-wide
/// (shared runner) — which is why `ids` is a list rather than the caller
/// hand-rolling a second teardown that can drift from this one.
fn reap(home: &Path, spt_bin: &Path, broker: &mut Child, brain_pid: u32, extra: &[u32], ids: &[&str]) {
    for p in extra {
        kill_pid(*p);
    }
    for id in ids {
        let psyche_perch =
            perch::resolve_perch_path(&format!("{id}-psyche"), ParentHint::Explicit(id));
        if let Some(p) = spt_store::info::read_pid(&psyche_perch) {
            kill_pid(p);
        }
    }
    let _ = {
        let mut cmd = Command::new(spt_bin);
        cmd.no_window().args(["daemon", "stop"]).env("SPT_HOME", home);
        common::output_bounded(cmd, Duration::from_secs(20))
    };
    kill_pid(brain_pid);
    let _ = broker.kill();
    let _ = broker.wait();
}

/// Poll an endpoint's on-disk status until `want`, bounded; returns whether seen.
fn wait_status(perch_path: &Path, want: &str, budget: Duration) -> bool {
    let deadline = Instant::now() + budget;
    while Instant::now() < deadline {
        if spt_store::info::read_info(perch_path).and_then(|i| i.status).as_deref() == Some(want) {
            return true;
        }
        std::thread::sleep(Duration::from_millis(120));
    }
    false
}

/// Extract the long-lived harness pid from the terse `ENDPOINT_RUN:… pid=NNN` line.
fn harness_pid_of(stderr: &str) -> Option<u32> {
    stderr
        .split("pid=")
        .nth(1)
        .and_then(|s| s.split_whitespace().next())
        .and_then(|s| s.trim().parse().ok())
}

/// Mint `<id>` under `<adapter>` if it does not exist yet, then
/// `spt endpoint start <id>` (headless), bounded; returns the START's Output.
fn run_start(home: &Path, spt_bin: &Path, adapter: &str, id: &str) -> Output {
    if !perch::resolve_perch_path(id, ParentHint::Infer).exists() {
        common::create_endpoint(spt_bin, home, id, adapter);
    }
    let mut cmd = Command::new(spt_bin);
    cmd.no_window()
        .args(["endpoint", "start", id])
        .env("SPT_HOME", home);
    common::output_bounded(cmd, Duration::from_secs(45))
}

/// `spt endpoint resume <id>` (latest-only since U3 — the session comes off the
/// ledger, not off the command line), bounded; returns the Output.
fn resume_latest(home: &Path, spt_bin: &Path, id: &str) -> Output {
    let mut cmd = Command::new(spt_bin);
    cmd.no_window()
        .args(["endpoint", "resume", id])
        .env("SPT_HOME", home);
    common::output_bounded(cmd, Duration::from_secs(45))
}

/// What an attached `spt rc <target>` streamed: `(stdout_captured, stderr,
/// saw_tick)`. Holds stdin open, reads stdout until the dummy heartbeat (or a
/// 12s deadline), then kills the viewport. `envs` land on the rc process.
fn rc_attach_capture(
    home: &Path,
    spt_bin: &Path,
    target: &str,
    envs: &[(&str, &str)],
) -> (String, String, bool) {
    let rc_err = home.join(format!("rc-{}.stderr.log", target.replace([':', '@'], "_")));
    let rc_err_file = std::fs::File::create(&rc_err).expect("create rc stderr log");
    let mut cmd = Command::new(spt_bin);
    cmd.no_window()
        .args(["rc", target])
        .env("SPT_HOME", home)
        // The single-pump counter (REQ-RC-SINGLE-PUMP-BRAIN) witnesses pump
        // CONSTRUCTION via PUMP_IPC_READER, which is SILENT by default now —
        // a client must not write diagnostics into a terminal it owns
        // (REQ-RC-DISPLAY-SOLE-WRITER). A rig opts in; its stderr is a file.
        // `envs` is applied AFTER, so a caller can still override.
        .env("SPT_PUMP_TRACE", "1")
        .stdout(Stdio::piped())
        .stderr(Stdio::from(rc_err_file))
        .stdin(Stdio::piped()); // hold stdin open — a closed stdin EOF-detaches early
    for (k, v) in envs {
        cmd.env(k, v);
    }
    let mut rc: Child = cmd.spawn().expect("spawn spt rc");
    let _rc_stdin = rc.stdin.take();
    let mut rc_out = rc.stdout.take().unwrap();
    let (tx, rx) = std::sync::mpsc::channel::<String>();
    std::thread::spawn(move || {
        let mut buf = [0u8; 4096];
        let mut acc = String::new();
        loop {
            match rc_out.read(&mut buf) {
                Ok(0) => break,
                Ok(n) => {
                    acc.push_str(&String::from_utf8_lossy(&buf[..n]));
                    if acc.contains("DUMMY_HARNESS_TICK") {
                        let _ = tx.send(acc.clone());
                        break;
                    }
                }
                Err(_) => break,
            }
        }
        let _ = tx.send(acc);
    });
    let captured = rx.recv_timeout(Duration::from_secs(12)).unwrap_or_default();
    let saw_tick = captured.contains("DUMMY_HARNESS_TICK");
    kill_pid(rc.id());
    let _ = rc.kill();
    let _ = rc.wait();
    let stderr = std::fs::read_to_string(&rc_err).unwrap_or_default();
    (captured, stderr, saw_tick)
}

// ── 1. offline row over an HONEST live session ⇒ rc attaches ─────────────────
//
// [int->REQ-RC-HONEST-SESSION-AUTHORITY] [int->REQ-HAZARD-RC-ATTACH-TRUTH]
// [int->REQ-RC-SINGLE-PUMP-BRAIN]
// The perri contradiction, reproduced organically: a live never-binding session
// (hold-unbound) whose perch is hand-stamped `offline` (the preserved field
// shape). Pre-fix `spt rc` trusted the row and refused pre-IPC; the honest-
// session gate must attach and stream the live heartbeat.
//
// This is the plain-bare-target path that runs the W1 honest-session probe
// (SessionProbe::connect) THEN establishes the attach — the exact invocation
// REQ-RC-SINGLE-PUMP-BRAIN governs: the probe's pump Brain must ride into
// establish_attach, not be discarded for a second cold_start_pump. Observable:
// `PUMP_IPC_READER: spawned` (brain.rs, once per pump construction) appears
// EXACTLY ONCE on rc's stderr, not twice.
#[test]
fn offline_row_over_live_session_attaches() {
    let _serial = E2E_LOCK.lock().unwrap_or_else(|e| e.into_inner());
    let home = tempfile::tempdir().unwrap();
    std::env::set_var("SPT_HOME", home.path());
    let spt_bin = PathBuf::from(env!("CARGO_BIN_EXE_spt"));
    let mock = common::sibling_bin("mock-session");
    assert!(mock.exists(), "build the dummy harness: cargo build -p mock-adapter --bin mock-session");

    seed_subnets(&["solo"]);
    register_harness(home.path(), &spt_bin, &mock, "holdharness", "hold-unbound");
    let (mut broker, brain_pid, brain_log) = spawn_broker(home.path(), &spt_bin);

    let id = "truth1";
    let run = run_start(home.path(), &spt_bin, "holdharness", id);
    let perch_path = perch::resolve_perch_path(id, ParentHint::Infer);
    let harness_pid = harness_pid_of(&String::from_utf8_lossy(&run.stderr));
    let was_unbound =
        wait_status(&perch_path, spt_store::liveness::STATUS_UNBOUND, Duration::from_secs(15));

    // Hand-stamp the organic contradiction: offline row over the live session.
    spt_store::info::set_status(&perch_path, spt_store::liveness::STATUS_OFFLINE).unwrap();

    let (captured, rc_stderr, saw_tick) = rc_attach_capture(home.path(), &spt_bin, id, &[]);
    let brain_stderr = common::daemon_stderr_panel(&brain_log);
    reap(home.path(), &spt_bin, &mut broker, brain_pid, harness_pid.as_slice(), &[id]);

    assert!(run.status.success(), "hold-unbound run: {}", String::from_utf8_lossy(&run.stderr));
    assert!(was_unbound, "precondition: the skeleton reads UNBOUND before the stamp.\n{brain_stderr}");
    assert!(
        !captured.contains("is offline — nothing to attach to"),
        "the offline fast-fail must NOT fire over an honest live session.\n=== rc stdout ===\n{captured}"
    );
    assert!(
        saw_tick,
        "rc must attach the live session the offline row lied about.\n\
         === rc stdout ===\n{captured}\n=== rc stderr ===\n{rc_stderr}\n=== brain ===\n{brain_stderr}"
    );
    // REQ-RC-SINGLE-PUMP-BRAIN: the probe's pump Brain rides into establish_attach
    // — one pump per invocation, not the probe's plus a second cold_start_pump.
    let pump_spawns = rc_stderr.matches("PUMP_IPC_READER: spawned").count();
    assert_eq!(
        pump_spawns, 1,
        "plain rc must construct EXACTLY ONE pump Brain (probe conn carried into \
         establish_attach); saw {pump_spawns}.\n=== rc stderr ===\n{rc_stderr}"
    );
}

// ── 2. offline row + NO session ⇒ the existing offline refusal ───────────────
//
// [int->REQ-RC-HONEST-SESSION-AUTHORITY] [int->REQ-HAZARD-RC-ATTACH-TRUTH]
// Absence corroborates the row: with no broker session, the offline fast-fail
// keeps its clean guidance (no hang, no attach).
#[test]
fn offline_row_with_no_session_refuses() {
    let _serial = E2E_LOCK.lock().unwrap_or_else(|e| e.into_inner());
    let home = tempfile::tempdir().unwrap();
    std::env::set_var("SPT_HOME", home.path());
    let spt_bin = PathBuf::from(env!("CARGO_BIN_EXE_spt"));
    let (mut broker, brain_pid, _brain_log) = spawn_broker(home.path(), &spt_bin);

    let id = "ghost1";
    let perch_path = perch::resolve_perch_path(id, ParentHint::Infer);
    std::fs::create_dir_all(&perch_path).unwrap();
    let rec = spt_store::info::InfoJson::new(id, "2026-07-18T00:00:00Z", 4242, "sid-x", "live_agent");
    spt_store::info::write_info(&perch_path, &rec).unwrap();
    spt_store::info::set_status(&perch_path, spt_store::liveness::STATUS_OFFLINE).unwrap();

    let out = {
        let mut cmd = Command::new(&spt_bin);
        cmd.no_window().args(["rc", id]).env("SPT_HOME", home.path());
        common::output_bounded(cmd, Duration::from_secs(20))
    };
    reap(home.path(), &spt_bin, &mut broker, brain_pid, &[], &[id]);

    let stdout = String::from_utf8_lossy(&out.stdout);
    assert!(out.status.success(), "the offline refusal is a clean exit");
    assert!(
        stdout.contains("is offline — nothing to attach to"),
        "offline + no session keeps the corroborated refusal.\n=== rc stdout ===\n{stdout}"
    );
}

// ── 3. claimed session, DEAD client tree ⇒ refusal, never an attach ──────────
//
// [int->REQ-RC-HONEST-SESSION-AUTHORITY] [int->REQ-HAZARD-RC-ATTACH-TRUTH]
// The zombie shape (W3's wedge): an adapter-labeled wrapper that SURVIVES its
// client chain's death via a childless builtin wait (cmd `pause` / sh `read`).
// The broker still claims the session; rc must refuse and never attach the
// corpse — and must not itself kill anything (reap belongs to the bringup verbs).
#[test]
fn zombie_claim_refuses_never_attaches() {
    let _serial = E2E_LOCK.lock().unwrap_or_else(|e| e.into_inner());
    let home = tempfile::tempdir().unwrap();
    std::env::set_var("SPT_HOME", home.path());
    let spt_bin = PathBuf::from(env!("CARGO_BIN_EXE_spt"));

    seed_subnets(&["solo"]);
    // The W2-surviving wrapper: client proxy (ping/sleep) dies, wrapper stays
    // alive in a CHILDLESS builtin wait — dead client chain, live root.
    #[cfg(windows)]
    let self_cmd = "cmd /c \"ping -n 3 127.0.0.1 >nul & pause\"";
    #[cfg(unix)]
    let self_cmd = "sh -c \"sleep 2; read x\"";
    register_raw_harness(home.path(), &spt_bin, "zombieharness", self_cmd, String::new());
    let (mut broker, brain_pid, brain_log) = spawn_broker(home.path(), &spt_bin);

    let id = "zomb1";
    let run = run_start(home.path(), &spt_bin, "zombieharness", id);
    let harness_pid = harness_pid_of(&String::from_utf8_lossy(&run.stderr));

    // Let the client proxy die (ping/sleep ~2s) — the wrapper stays.
    std::thread::sleep(Duration::from_secs(4));
    let wrapper_alive_pre = harness_pid.is_some_and(spt_store::proc::is_process_alive);

    let out = {
        let mut cmd = Command::new(&spt_bin);
        cmd.no_window()
            .args(["rc", id])
            .env("SPT_HOME", home.path())
            // Past-grace verdict without the 30s default wait.
            .env("SPT_SPAWN_CLIENT_GRACE_MS", "500");
        common::output_bounded(cmd, Duration::from_secs(20))
    };
    let wrapper_alive_post = harness_pid.is_some_and(spt_store::proc::is_process_alive);
    let brain_stderr = common::daemon_stderr_panel(&brain_log);
    reap(home.path(), &spt_bin, &mut broker, brain_pid, harness_pid.as_slice(), &[id]);

    assert!(run.status.success(), "zombie-shape run: {}", String::from_utf8_lossy(&run.stderr));
    assert!(wrapper_alive_pre, "precondition: the wrapper survives its client chain.\n{brain_stderr}");
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert!(
        stdout.contains("defunct session"),
        "rc must name the dead-client-tree claim, never attach it.\n=== rc stdout ===\n{stdout}"
    );
    assert!(
        !stdout.contains("DUMMY_HARNESS_TICK"),
        "a zombie claim must never stream an attach.\n{stdout}"
    );
    assert!(
        wrapper_alive_post,
        "rc REFUSES — it never reaps (reap belongs to the bringup dup-guard)"
    );
}

// ── 4. resume-never-bound reads UNBOUND and is attachable ────────────────────
//
// [int->REQ-RESUME-UNBOUND-STAMP] [int->REQ-HAZARD-RC-ATTACH-TRUTH]
// The writer-truth complement: bring an endpoint ONLINE (dummy binds), stop it
// (offline), then RESUME it with a hold-unbound harness (the stuck-native-
// resume field shape: session lives, bind never fires). Pre-fix the row stayed
// `offline` through the whole window; the resume launch must stamp UNBOUND,
// and the live session must be attachable.
#[test]
fn resume_never_bound_reads_unbound_and_attaches() {
    let _serial = E2E_LOCK.lock().unwrap_or_else(|e| e.into_inner());
    let home = tempfile::tempdir().unwrap();
    std::env::set_var("SPT_HOME", home.path());
    let spt_bin = PathBuf::from(env!("CARGO_BIN_EXE_spt"));
    let mock = common::sibling_bin("mock-session");
    assert!(mock.exists(), "build the dummy harness");

    seed_subnets(&["solo"]);
    register_harness(home.path(), &spt_bin, &mock, "dummyharness", "dummy");
    let (mut broker, brain_pid, brain_log) = spawn_broker(home.path(), &spt_bin);

    let id = "res1";
    let perch_path = perch::resolve_perch_path(id, ParentHint::Infer);
    let run1 = run_start(home.path(), &spt_bin, "dummyharness", id);
    let pid1 = harness_pid_of(&String::from_utf8_lossy(&run1.stderr));
    let online = wait_status(&perch_path, spt_store::liveness::STATUS_ONLINE, Duration::from_secs(20));

    // Kill the harness (the natural death path — the field shape's launchpad):
    // the broker's death observers reap the session and terminal-normalize the
    // row offline.
    if let Some(p) = pid1 {
        kill_pid(p);
    }
    let offline = wait_status(&perch_path, spt_store::liveness::STATUS_OFFLINE, Duration::from_secs(20));

    // RESUME into a never-binding harness: the pre-bind window is permanent.
    //
    // Since U3 the resume verb takes no adapter — it runs under the one the
    // endpoint recorded — so the never-binding behaviour is installed by
    // RE-REGISTERING that same adapter in hold-unbound mode. That is closer to
    // the field shape this row is about (a harness whose own native resume
    // hangs) than swapping to a differently-named adapter ever was.
    register_harness(home.path(), &spt_bin, &mock, "dummyharness", "hold-unbound");
    let run2 = resume_latest(home.path(), &spt_bin, id);
    let pid2 = harness_pid_of(&String::from_utf8_lossy(&run2.stderr));
    // THE writer-truth assert: the resumed-but-never-bound perch reads UNBOUND,
    // not offline (pre-fix: stayed offline forever).
    let unbound = wait_status(&perch_path, spt_store::liveness::STATUS_UNBOUND, Duration::from_secs(15));

    let (captured, rc_stderr, saw_tick) = rc_attach_capture(home.path(), &spt_bin, id, &[]);
    let brain_stderr = common::daemon_stderr_panel(&brain_log);
    let extra: Vec<u32> = pid1.into_iter().chain(pid2).collect();
    reap(home.path(), &spt_bin, &mut broker, brain_pid, &extra, &[id]);

    assert!(run1.status.success(), "bind bringup: {}", String::from_utf8_lossy(&run1.stderr));
    assert!(online, "precondition: the dummy binds ONLINE.\n{brain_stderr}");
    assert!(offline, "precondition: the harness death lands the offline row.\n{brain_stderr}");
    assert!(run2.status.success(), "resume launch: {}", String::from_utf8_lossy(&run2.stderr));
    assert!(
        unbound,
        "a resuming perch must read UNBOUND (not offline) through the pre-bind window.\n\
         === run2 stderr ===\n{}\n=== perch ===\n{:?}\n=== brain ===\n{brain_stderr}",
        String::from_utf8_lossy(&run2.stderr),
        spt_store::info::read_info(&perch_path)
    );
    assert!(
        saw_tick,
        "the resumed live session must be attachable.\n\
         === rc stdout ===\n{captured}\n=== rc stderr ===\n{rc_stderr}\n=== brain ===\n{brain_stderr}"
    );
}

// ── 5. harness-only ⇒ truthful pre-stream refusal (local row + gossiped) ─────
//
// [int->REQ-RC-HARNESS-ONLY-REFUSAL] [int->REQ-HAZARD-RC-ATTACH-TRUTH]
// Both carriers of the already-known fact: a LOCAL online live_agent row with
// `controllable=Some(false)`, and a gossiped remote `Instance{harness_only}`.
// Each refuses BEFORE any attach/stream naming the harness-hosted state —
// never the stale-row guess ("may have ended / registry row is stale").
#[test]
fn harness_only_refuses_truthfully_pre_stream() {
    let _serial = E2E_LOCK.lock().unwrap_or_else(|e| e.into_inner());
    let home = tempfile::tempdir().unwrap();
    std::env::set_var("SPT_HOME", home.path());
    let spt_bin = PathBuf::from(env!("CARGO_BIN_EXE_spt"));
    // This fixture injects a snapshot, not a peer heartbeat. A live pump may
    // hydrate it and correctly evict the unheard foreign node before rc reads.
    std::fs::write(
        home.path().join("daemon.json"),
        br#"{"relay":"disabled","peer_pump":false}"#,
    )
    .unwrap();
    let (mut broker, brain_pid, _brain_log) = spawn_broker(home.path(), &spt_bin);

    // LOCAL: the reproduced ONLINE - HARNESS ONLY row (api-listen shape).
    let local_id = "hh1";
    let perch_path = perch::resolve_perch_path(local_id, ParentHint::Infer);
    std::fs::create_dir_all(&perch_path).unwrap();
    let mut rec =
        spt_store::info::InfoJson::new(local_id, "2026-07-18T00:00:00Z", 4242, "sid-h", "live_agent");
    rec.controllable = Some(false);
    spt_store::info::write_info(&perch_path, &rec).unwrap();
    spt_store::info::set_status(&perch_path, spt_store::liveness::STATUS_ONLINE).unwrap();

    let local_out = {
        let mut cmd = Command::new(&spt_bin);
        cmd.no_window().args(["rc", local_id]).env("SPT_HOME", home.path());
        common::output_bounded(cmd, Duration::from_secs(20))
    };

    // REMOTE: a gossiped Instance{harness_only:true} on a foreign node.
    let remote_id = "hh2";
    let mut reg = spt_net::net::registry::SubnetRegistry::new();
    reg.merge_instance(
        remote_id,
        spt_net::net::registry::Instance {
            node: "aaaa1111deadbeefaaaa1111deadbeefaaaa1111deadbeefaaaa1111deadbeef".to_string(),
            status: spt_net::net::registry::Status::Active,
            epoch: 1,
            resources: None,
            last_active_ms: None,
            shell_adapters: Vec::new(),
            node_label: None,
            machine_id: None,
            endpoint_type: None,
            bound: true,
            controller_node: None,
            harness_only: true,
            adapter: None,
            recent_projects: Vec::new(),
            controlled: false,
        },
    );
    let reg_dir = perch::identity_dir().join("registry");
    std::fs::create_dir_all(&reg_dir).unwrap();
    std::fs::write(reg_dir.join("solo.json"), serde_json::to_string(&reg).unwrap()).unwrap();

    let remote_out = {
        let mut cmd = Command::new(&spt_bin);
        cmd.no_window().args(["rc", remote_id]).env("SPT_HOME", home.path());
        common::output_bounded(cmd, Duration::from_secs(20))
    };
    reap(home.path(), &spt_bin, &mut broker, brain_pid, &[], &[local_id]);

    for (label, out) in [("local", &local_out), ("remote", &remote_out)] {
        let stdout = String::from_utf8_lossy(&out.stdout);
        assert!(out.status.success(), "{label}: the preflight refusal is a clean exit");
        assert!(
            stdout.contains("harness-hosted"),
            "{label}: the refusal names the actual state.\n=== rc stdout ===\n{stdout}"
        );
        assert!(
            !stdout.contains("may have ended") && !stdout.contains("registry row is stale"),
            "{label}: never the stale-row guess.\n{stdout}"
        );
        assert!(
            !stdout.contains("is offline — nothing to attach to"),
            "{label}: not the offline copy either — the endpoint is ONLINE.\n{stdout}"
        );
    }
}


// ── 6. rc owns its display: NO pump diagnostic interleaves it ───────────────
//
// [int->REQ-RC-DISPLAY-SOLE-WRITER]
// hertz RCA 3 bug 1: `PUMP_IPC_READER: spawned` (a BrainConn reader-thread
// diagnostic) rendered INSIDE rc's owned surface — the reconnect banner homes
// the cursor and leaves it parked with no trailing newline, so the operator saw
// `Reconnecting to local daemon... 9sPUMP_IPC_READER: spawned`, and the field
// screenshot showed it reaching a real harness alt screen.
//
// WHY AN INT, AND WHY A UNIT WOULD BE A LIE (the REQ says so explicitly): the
// defect is an out-of-band write from a BACKGROUND THREAD to the process's
// inherited stderr. Those bytes never pass through the renderer, so no
// banner-byte unit can observe them. The honest oracle is a REAL `spt rc`
// process with stdout and stderr captured into ONE sink — which is precisely
// what a terminal is.
//
// THE OBSERVABILITY PROBLEM AND ITS ANSWER — a NAMED SIBLING PROBE. An
// absence assertion is worthless unless the rig can be shown capable of
// observing the presence. So the identical scenario runs TWICE:
//   probe: SPT_PUMP_TRACE=1 ⇒ the marker MUST appear (the oracle can see);
//   default: no env        ⇒ the marker MUST NOT appear (the product is silent).
// The probe run is what makes the default run's silence mean something. Without
// it, a rig that merely failed to capture stderr would pass forever.
#[test]
fn rc_owned_display_carries_no_pump_diagnostics() {
    let _serial = E2E_LOCK.lock().unwrap_or_else(|e| e.into_inner());
    let home = tempfile::tempdir().unwrap();
    std::env::set_var("SPT_HOME", home.path());
    let spt_bin = PathBuf::from(env!("CARGO_BIN_EXE_spt"));
    let mock = common::sibling_bin("mock-session");
    assert!(mock.exists(), "build the dummy harness: cargo build -p mock-adapter --bin mock-session");

    seed_subnets(&["solo"]);
    register_harness(home.path(), &spt_bin, &mock, "soleharness", "hold-unbound");
    let (mut broker, brain_pid, brain_log) = spawn_broker(home.path(), &spt_bin);

    // One rc attach, stdout+stderr into ONE sink (the terminal, modeled).
    // Returns everything the operator would have seen.
    let attach_and_capture = |id: &str, trace: bool| -> String {
        let sink_path = home.path().join(format!("rc-combined-{id}.log"));
        let sink = std::fs::File::create(&sink_path).expect("create combined sink");
        let sink_err = sink.try_clone().expect("clone the sink handle");
        let mut cmd = Command::new(&spt_bin);
        cmd.no_window()
            .args(["rc", id])
            .env("SPT_HOME", home.path())
            .stdout(Stdio::from(sink))
            .stderr(Stdio::from(sink_err))
            .stdin(Stdio::piped());
        if trace {
            cmd.env("SPT_PUMP_TRACE", "1");
        }
        let mut rc: Child = cmd.spawn().expect("spawn spt rc");
        let _rc_stdin = rc.stdin.take();
        let read_sink = || std::fs::read_to_string(&sink_path).unwrap_or_default();
        let deadline = Instant::now() + Duration::from_secs(30);
        while Instant::now() < deadline && !read_sink().contains("DUMMY_HARNESS_TICK") {
            std::thread::sleep(Duration::from_millis(150));
        }
        let seen = read_sink();
        kill_pid(rc.id());
        let _ = rc.kill();
        let _ = rc.wait();
        seen
    };

    // ONE ENDPOINT PER LEG, deliberately. `attach_and_capture` kills rc and
    // returns WITHOUT waiting for that rc's session/controller release, so two
    // sequential attaches against a SINGLE endpoint make the second race the
    // first's teardown inside a fixed 30s budget — observed red under load, and
    // the precondition below then (correctly) refuses to judge a state the rig
    // never built. Two ids REMOVE the contention rather than waiting it out
    // (determinism over a convergence poll: a bounded wait only lowers the rate,
    // and a lower rate on a race moves the red onto someone else's PR).
    // The probe's meaning survives the split: it rests on an identical CODE PATH
    // and SETUP, not on identical endpoint identity — the sole-writer property
    // is a property of the rc process, not of which endpoint it attached to.
    let probe_id = "sole-probe";
    let default_id = "sole-default";
    let probe_run = run_start(home.path(), &spt_bin, "soleharness", probe_id);
    let default_start = run_start(home.path(), &spt_bin, "soleharness", default_id);
    let mut harness_pids: Vec<u32> =
        harness_pid_of(&String::from_utf8_lossy(&probe_run.stderr)).into_iter().collect();
    harness_pids.extend(harness_pid_of(&String::from_utf8_lossy(&default_start.stderr)));

    let probe = attach_and_capture(probe_id, true);
    let default_run = attach_and_capture(default_id, false);

    reap(home.path(), &spt_bin, &mut broker, brain_pid, &harness_pids, &[probe_id, default_id]);
    let brain_stderr = common::daemon_stderr_panel(&brain_log);

    for (label, run) in [("probe", &probe_run), ("default", &default_start)] {
        assert!(
            run.status.success(),
            "{label} hold-unbound run: {}",
            String::from_utf8_lossy(&run.stderr)
        );
    }
    // Non-vacuity 1: both runs really attached — an rc that never connected
    // would carry no marker for uninteresting reasons.
    assert!(
        probe.contains("DUMMY_HARNESS_TICK") && default_run.contains("DUMMY_HARNESS_TICK"),
        "precondition: both attaches streamed.\n=== probe ===\n{probe}\n\
         === default ===\n{default_run}\n=== brain ===\n{brain_stderr}"
    );
    // Non-vacuity 2 — THE SIBLING PROBE: with the valve open the marker DOES
    // reach this sink, so the sink observes the very bytes the next assertion
    // denies. This is the assertion that makes the absence below meaningful.
    assert!(
        probe.contains("PUMP_IPC_READER"),
        "the rig must be able to SEE the marker (SPT_PUMP_TRACE=1) — otherwise \
         the absence assertion below proves nothing.\n=== probe ===\n{probe}"
    );
    // THE SUBJECT: by default rc's owned display carries no pump diagnostics.
    assert!(
        !default_run.contains("PUMP_IPC_READER"),
        "an ordinary `spt rc` must write NO pump diagnostics into the display it \
         owns (the field corruption: `Reconnecting to local daemon... \
         9sPUMP_IPC_READER: spawned`).\n=== default ===\n{default_run}"
    );
}