fux 0.12.0

use serde_json::{Value, json};
use std::{
    fs,
    path::{Path, PathBuf},
    process::{Child, Command, Stdio},
    sync::atomic::{AtomicU64, Ordering},
    thread,
    time::{Duration, Instant},
};

static NEXT: AtomicU64 = AtomicU64::new(0);
// Serialize server spawns with outer-PTY spawns, so a concurrent fork never
// inherits a descriptor another fixture is about to hand to its child.
static SPAWN: std::sync::Mutex<()> = std::sync::Mutex::new(());

mod brp;
mod design;
mod socket;

type Fail = Box<dyn std::error::Error>;
#[path = "../src/testing.rs"]
mod testing;
#[path = "../src/unix_http.rs"]
mod unix_http;
use testing::{Need, Outcome, ScreenText};

/// JSON lookup with `Value`'s own null-on-miss semantics, without the index lint.
trait At {
    fn at<I: serde_json::value::Index>(&self, index: I) -> Value;
}
impl At for Value {
    fn at<I: serde_json::value::Index>(&self, index: I) -> Value {
        self.get(index).cloned().unwrap_or(Value::Null)
    }
}
/// Iterate a JSON array; anything else is an empty set of rows.
trait Rows {
    fn rows(&self) -> impl Iterator<Item = &Value>;
}
impl Rows for Value {
    fn rows(&self) -> impl Iterator<Item = &Value> {
        self.as_array().into_iter().flatten()
    }
}

struct Server {
    child: Child,
    /// The server's socket, in a private directory of this fixture.
    socket: PathBuf,
    directory: PathBuf,
}

/// A fixture directory under the system temporary directory, and the socket
/// path inside its private `s/` subdirectory, which the server creates.
fn fixture() -> Result<(PathBuf, PathBuf), Fail> {
    let directory = std::env::temp_dir().join(format!(
        "fux-test-{}-{}",
        std::process::id(),
        NEXT.fetch_add(1, Ordering::Relaxed)
    ));
    fs::create_dir(&directory)?;
    let directory = directory.canonicalize()?;
    let socket = directory.join("s").join("fux.sock");
    Ok((directory, socket))
}

/// A server command with the fixture environment: the configured shell and
/// a clean environment for the transport.
fn server_command(directory: &Path, args: &[&std::ffi::OsStr]) -> Command {
    let mut command = Command::new(env!("CARGO_BIN_EXE_fux"));
    command
        .arg("server")
        .args(args)
        .env("SHELL", "/bin/sh")
        .env("PS1", "$ ")
        .env("HOME", directory)
        .env("HISTFILE", "/dev/null")
        .env_remove("ENV")
        .env_remove("BASH_ENV")
        .env_remove("FUX_ENDPOINT")
        .env_remove("FUX_SOCKET")
        .current_dir(directory);
    command
}

impl Server {
    fn start() -> Result<Self, Fail> {
        let _spawn = SPAWN.lock().unwrap_or_else(|error| error.into_inner());
        let (directory, socket) = fixture()?;
        let config = directory.join("fux.json");
        fs::write(
            &config,
            r#"{"shell":["/bin/sh"],"history_lines":100,"clipboard":"write-only"}"#,
        )?;
        let child = server_command(
            &directory,
            &[
                "--socket".as_ref(),
                socket.as_ref(),
                "--config".as_ref(),
                config.as_ref(),
            ],
        )
        .stdout(Stdio::null())
        .stderr(fs::File::create(directory.join("server.log"))?)
        .spawn()?;
        let server = Self {
            child,
            socket,
            directory,
        };
        eventually(|| Ok(server.request("rpc.discover", Value::Null).is_ok()))?;
        Ok(server)
    }

    fn request(&self, method: &str, params: Value) -> Result<Value, String> {
        let response: Value = unix_http::agent(&self.socket, Some(Duration::from_secs(5)))
            .post(unix_http::URL)
            .send_json(json!({"jsonrpc":"2.0","id":1,"method":method,"params":params}))
            .map_err(|error| error.to_string())?
            .body_mut()
            .read_json()
            .map_err(|error| error.to_string())?;
        if let Some(error) = response.get("error") {
            return Err(error.to_string());
        }
        Ok(response.at("result"))
    }

    fn rpc(&self, method: &str, params: Value) -> Result<Value, String> {
        self.request(method, params)
            .map_err(|error| format!("{method}: {error}"))
    }

    fn query(&self, component: &str) -> Result<Value, String> {
        let rows = self.rpc("world.query", json!({"data":{"components":[component]}}))?;
        if !rows.is_array() {
            return Err(format!("world.query {component} did not return rows"));
        }
        Ok(rows)
    }

    /// Sends one `Command` as JSON; see `control.rs` for the shapes.
    fn control(&self, viewer: u64, command: Value) -> Result<(), String> {
        self.rpc(
            "world.trigger_event",
            json!({"event":"fux::control::Control","value":{"viewer":viewer,"command":command}}),
        )?;
        Ok(())
    }
    /// A command that is only its kind.
    fn command(&self, viewer: u64, kind: &str) -> Result<(), String> {
        self.control(viewer, json!({"kind":kind}))
    }
    fn scoped(&self, viewer: u64, kind: &str, scope: &str) -> Result<(), String> {
        self.control(viewer, json!({"kind":kind,"scope":scope}))
    }
    fn split(&self, viewer: u64, axis: &str, program: Option<&str>) -> Result<(), String> {
        self.control(
            viewer,
            json!({"kind":"split","axis":axis,"program":program}),
        )
    }
    fn focus(&self, viewer: u64, pane: u64) -> Result<(), String> {
        self.control(viewer, json!({"kind":"focus","pane":pane}))
    }
    fn tab_new(&self, viewer: u64, name: Option<&str>) -> Result<(), String> {
        self.control(viewer, json!({"kind":"tab_new","name":name}))
    }
    fn close(&self, viewer: u64, subject: Value) -> Result<(), String> {
        self.control(viewer, json!({"kind":"close","subject":subject}))
    }
    /// One relationship component of a viewer, or null when it has none.
    fn relation(&self, viewer: u64, component: &str) -> Result<Value, String> {
        Ok(self
            .query(component)?
            .rows()
            .find(|row| row.at("entity") == viewer)
            .map_or(Value::Null, |row| row.at("components").at(component)))
    }
    fn viewing(&self, viewer: u64) -> Result<Value, String> {
        self.relation(viewer, "fux::model::Viewing")
    }
    fn on_tab(&self, viewer: u64) -> Result<Value, String> {
        self.relation(viewer, "fux::model::OnTab")
    }
    fn focused(&self, viewer: u64) -> Result<Value, String> {
        self.relation(viewer, "fux::model::Focused")
    }
    /// The workspace this viewer is looking at.
    fn workspace_of(&self, viewer: u64) -> Result<u64, String> {
        self.viewing(viewer)?
            .as_u64()
            .ok_or_else(|| "viewer has no workspace".into())
    }

    fn input(&self, viewer: u64, input: Value) -> Result<(), String> {
        self.rpc(
            "world.trigger_event",
            json!({"event":"fux::control::UserInput","value":{"viewer":viewer,"input":input}}),
        )?;
        Ok(())
    }

    fn enter(&self, viewer: u64) -> Result<(), String> {
        self.input(
            viewer,
            json!({"kind":"key","key":"enter","ctrl":false,"alt":false,"shift":false}),
        )
    }

    fn attach(&self) -> Result<u64, String> {
        self.rpc("fux.attach", json!({"rows":24,"cols":80}))?
            .at("viewer")
            .as_u64()
            .ok_or_else(|| "fux.attach did not return a viewer".into())
    }

    fn screen(&self, viewer: u64) -> Result<String, String> {
        let frame = self.rpc("fux.frame", json!({"viewer":viewer}))?;
        let paint = frame.at("paint");
        let mut parser = fux_vt::Parser::new(24, 80, 0).need()?;
        parser
            .process(paint.as_str().ok_or("frame has no paint")?.as_bytes())
            .need()?;
        Ok(parser.screen().contents())
    }
}

impl Drop for Server {
    fn drop(&mut self) {
        let _ = self.request(
            "world.trigger_event",
            json!({"event":"fux::control::Shutdown","value":null}),
        );
        let deadline = Instant::now() + Duration::from_secs(5);
        while matches!(self.child.try_wait(), Ok(None)) && Instant::now() < deadline {
            thread::sleep(Duration::from_millis(10));
        }
        let _ = self.child.kill();
        let _ = self.child.wait();
        if thread::panicking() {
            eprintln!(
                "server log: {}",
                fs::read_to_string(self.directory.join("server.log")).unwrap_or_default()
            );
        }
        let _ = fs::remove_dir_all(&self.directory);
    }
}

/// Polls until the observation holds; an observation error or timeout is the failure.
#[track_caller]
fn eventually(mut observed: impl FnMut() -> Result<bool, Fail>) -> Result<(), Fail> {
    let location = std::panic::Location::caller();
    let deadline = Instant::now() + Duration::from_secs(5);
    while !observed()? {
        if Instant::now() >= deadline {
            return Err(format!(
                "observable state did not settle within five seconds at {location}"
            )
            .into());
        }
        thread::sleep(Duration::from_millis(10));
    }
    Ok(())
}

fn alive(pid: i32) -> bool {
    nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None).is_ok()
}

#[test]
fn stock_launch_removal_settles_without_another_request() -> Outcome {
    let server = Server::start()?;
    let pid_file = server.directory.join("child.pid");
    let entity = server.rpc("world.spawn_entity", json!({"components":{
        "fux::model::Launch":{"argv":["/bin/sh","-c",format!("echo $$ > '{}'; exec sleep 60",pid_file.display())],"cwd":server.directory,"history_lines":20}
    }}))?.at("entity").as_u64().need()?;
    // Observe the OS, not another RPC: a later request would mask a missing wake.
    eventually(|| Ok(fs::read_to_string(&pid_file).is_ok_and(|text| !text.trim().is_empty())))?;
    let pid = fs::read_to_string(pid_file)?.trim().parse()?;
    assert!(alive(pid));
    server.rpc(
        "world.remove_components",
        json!({"entity":entity,"components":["fux::model::Launch"]}),
    )?;
    eventually(|| Ok(!alive(pid)))?;
    let states = server.query("fux::model::ProcessState")?;
    let state = &states
        .rows()
        .find(|row| row.at("entity") == entity)
        .need()?
        .at("components")
        .at("fux::model::ProcessState");
    assert!(state.at("status").at("pid").is_null());
    assert_eq!(state.at("status").at("kind"), "exited", "{state}");
    assert!(state.at("status").at("code").is_number(), "{state}");
    Ok(())
}

/// F1 of the agent exercises: a `Viewer` written back without `notice` killed
/// the server. A partial payload is now a typed rejection naming the missing
/// field, or, for the one component that keeps its reflect encoding, its
/// reflected default. An `Option` field the published schema does not list as
/// required is accepted as null. Nothing here may end the server.
#[test]
fn partial_component_payloads_are_rejected_without_ending_the_server() -> Outcome {
    let server = Server::start()?;
    let viewer = server.attach()?;
    let insert = |entity: u64, component: &str, value: Value| {
        server.rpc(
            "world.insert_components",
            json!({"entity":entity,"components":{component:value}}),
        )
    };
    let rejected = |entity: u64, component: &str, value: Value, field: &str| -> Outcome {
        let error = insert(entity, component, value).err().need()?;
        assert!(
            error.contains(&format!("missing field `{field}`")),
            "{component}: {error}"
        );
        Ok(())
    };
    // The exact payload from `evidence/partial-component-panic.sh`.
    insert(
        viewer,
        "fux::model::Viewer",
        json!({"rows":24,"cols":80,"zoom":false,"scrollback":5}),
    )?;
    let state = server.relation(viewer, "fux::model::Viewer")?;
    assert_eq!(state.at("scrollback"), 5, "{state}");
    assert!(state.at("notice").is_null(), "{state}");
    rejected(
        viewer,
        "fux::model::Viewer",
        json!({"cols":80,"zoom":false,"scrollback":0,"notice":null}),
        "rows",
    )?;
    rejected(viewer, "fux::interaction::Prefix", json!({}), "scroll")?;

    let pid_file = server.directory.join("child.pid");
    let argv = json!([
        "/bin/sh",
        "-c",
        format!("echo $$ > '{}'; exec sleep 60", pid_file.display())
    ]);
    let pane = server
        .rpc(
            "world.spawn_entity",
            json!({"components":{
                "fux::model::Launch":{"argv":argv,"cwd":server.directory,"history_lines":20}
            }}),
        )?
        .at("entity")
        .as_u64()
        .need()?;
    eventually(|| Ok(fs::read_to_string(&pid_file).is_ok_and(|text| !text.trim().is_empty())))?;
    let pid: i32 = fs::read_to_string(&pid_file)?.trim().parse()?;
    rejected(
        pane,
        "fux::model::Launch",
        json!({"argv":["/bin/sh"]}),
        "cwd",
    )?;
    // Writing the same recipe back replaces the recipe, not the process: a
    // replaced component is a change, not an addition, so nothing respawns.
    insert(
        pane,
        "fux::model::Launch",
        json!({"argv":argv,"cwd":server.directory,"history_lines":20}),
    )?;
    let process = |pane: u64| -> Result<Value, String> {
        Ok(server
            .query("fux::model::ProcessState")?
            .rows()
            .find(|row| row.at("entity") == pane)
            .map_or(Value::Null, |row| {
                row.at("components").at("fux::model::ProcessState")
            }))
    };
    for _ in 0..20 {
        thread::sleep(Duration::from_millis(25));
        let state = process(pane)?;
        assert_eq!(state.at("status").at("kind"), "running", "{state}");
        assert_eq!(state.at("status").at("pid"), pid, "{state}");
    }
    assert!(alive(pid));
    assert_eq!(fs::read_to_string(&pid_file)?.trim().parse::<i32>()?, pid);

    let view = server
        .rpc(
            "world.spawn_entity",
            json!({"components":{"fux::model::PaneView":{"pane":pane}}}),
        )?
        .at("entity")
        .as_u64()
        .need()?;
    rejected(view, "fux::model::PaneView", json!({}), "pane")?;
    let error = server
        .rpc(
            "world.spawn_entity",
            json!({"components":{"fux::model::PaneView":{}}}),
        )
        .err()
        .need()?;
    assert!(error.contains("missing field `pane`"), "{error}");

    // `Overlay` keeps its reflect encoding and falls back to its default: an
    // empty list whose placeholder target execution validates like a capture.
    insert(viewer, "fux::interaction::Overlay", json!({}))?;
    assert!(!server.screen(viewer)?.is_empty());
    server.enter(viewer)?;
    server.input(
        viewer,
        json!({"kind":"key","key":"escape","ctrl":false,"alt":false,"shift":false}),
    )?;
    eventually(|| {
        Ok(server
            .relation(viewer, "fux::interaction::Overlay")?
            .is_null())
    })?;

    server.rpc("rpc.discover", Value::Null)?;
    assert!(!server.screen(viewer)?.is_empty());
    assert!(alive(pid));
    Ok(())
}

#[test]
fn interactive_background_jobs_hang_up_when_pane_terminates() -> Outcome {
    let server = Server::start()?;
    let viewer = server.attach()?;
    let shell = server
        .query("fux::model::ProcessState")?
        .at(0)
        .at("components")
        .at("fux::model::ProcessState")
        .at("status")
        .at("pid")
        .as_i64()
        .need()? as i32;
    let pid_file = server.directory.join("background.pid");
    server.input(
        viewer,
        json!({"kind":"paste","text":format!("sleep 60 & echo $! > '{}'",pid_file.display())}),
    )?;
    server.enter(viewer)?;
    eventually(|| Ok(fs::read_to_string(&pid_file).is_ok_and(|text| !text.trim().is_empty())))?;
    let background: i32 = fs::read_to_string(pid_file)?.trim().parse()?;
    assert_ne!(
        nix::unistd::getpgid(Some(nix::unistd::Pid::from_raw(background)))?.as_raw(),
        shell
    );
    server.command(viewer, "terminate")?;
    eventually(|| Ok(!alive(shell) && !alive(background)))?;
    let state = &server
        .query("fux::model::ProcessState")?
        .at(0)
        .at("components")
        .at("fux::model::ProcessState");
    assert!(state.at("status").at("pid").is_null());
    assert_eq!(state.at("status").at("kind"), "exited", "{state}");
    assert!(state.at("status").at("code").is_number(), "{state}");
    Ok(())
}

#[test]
fn layout_mapping_and_prompt_paste_preserve_live_process_identity() -> Outcome {
    let server = Server::start()?;
    let viewer = server.attach()?;
    let first = server
        .query("fux::model::Launch")?
        .at(0)
        .at("entity")
        .as_u64()
        .need()?;
    server.command(viewer, "zoom")?;
    server.control(viewer, json!({"kind":"scroll","order":"previous"}))?;
    server.split(viewer, "horizontal", None)?;
    let screen = server.screen(viewer)?;
    let chrome = screen.lines().last().need()?;
    assert!(!chrome.contains("zoom"));
    assert!(!chrome.contains('↑'));
    let launches = server.query("fux::model::Launch")?;
    let second = launches
        .rows()
        .filter_map(|row| row.at("entity").as_u64())
        .find(|entity| *entity != first)
        .need()?;
    for (entity, name) in [(first, "alpha"), (second, "beta")] {
        server.rpc(
            "world.insert_components",
            json!({"entity":entity,"components":{"bevy_ecs::name::Name":name}}),
        )?;
    }
    // Distinct real terminal contents, not decorative pane-header labels.
    for marker in ["BETA", "ALPHA"] {
        server.input(
            viewer,
            json!({"kind":"paste","text":format!("printf '\\033[2J\\033[H{marker}\\n'")}),
        )?;
        server.enter(viewer)?;
        eventually(|| Ok(server.screen(viewer)?.contains(marker)))?;
        server.command(viewer, "focus_next")?;
        server.screen(viewer)?;
    }
    let before = server.query("fux::model::ProcessState")?;
    let scene = server.directory.join("layout.scn.ron");
    let workspace = server.workspace_of(viewer)?;
    server.control(
        viewer,
        json!({"kind":"save_layout","workspace":workspace,"path":scene}),
    )?;
    eventually(|| Ok(fs::metadata(&scene).is_ok_and(|metadata| metadata.len() > 0)))?;
    server.control(viewer, json!({
        "kind":"load_layout","workspace":workspace,"path":scene,"mapping":[[first,second],[second,first]]
    }))?;
    eventually(|| Ok(server.screen(viewer)?.contains("loaded ")))?;
    let screen = server.screen(viewer)?;
    let content = screen.lines().next().need()?;
    assert!(
        content.find("BETA").need()? < content.find("ALPHA").need()?,
        "{screen}"
    );
    let after = server.query("fux::model::ProcessState")?;
    for old in before.rows() {
        let new = after
            .rows()
            .find(|row| row.at("entity") == old.at("entity"))
            .need()?;
        assert_eq!(
            old.at("components")
                .at("fux::model::ProcessState")
                .at("status")
                .at("pid"),
            new.at("components")
                .at("fux::model::ProcessState")
                .at("status")
                .at("pid")
        );
    }
    let focus = server.focused(viewer)?.as_u64().need()?;
    server.rpc(
        "world.insert_components",
        json!({"entity":focus,"components":{"bevy_camera::visibility::Visibility":"Hidden"}}),
    )?;
    eventually(|| Ok(!server.screen(viewer)?.contains("BETA")))?;
    server.rpc(
        "world.remove_components",
        json!({"entity":focus,"components":["bevy_camera::visibility::Visibility"]}),
    )?;
    eventually(|| Ok(server.screen(viewer)?.contains("BETA")))?;
    // The first load replaced the workspace entity; a command names the live one.
    let workspace = server.workspace_of(viewer)?;
    server.control(
        viewer,
        json!({"kind":"load_layout","workspace":workspace,"path":scene,"mapping":[[first,first+1_000_000]]}),
    )?;
    eventually(|| Ok(server.screen(viewer)?.contains("missing live pane")))?;
    let remaining = server.query("fux::model::PaneView")?;
    assert!(remaining.rows().any(|row| row.at("entity") == focus));
    assert_eq!(remaining.rows().count(), 2);
    Ok(())
}

#[test]
fn stock_viewer_removal_releases_its_native_size_constraint() -> Outcome {
    let server = Server::start()?;
    let viewer = server.attach()?;
    for despawn in [false, true] {
        let small = server
            .rpc("fux.attach", json!({"rows":12,"cols":40}))?
            .at("viewer")
            .as_u64()
            .need()?;
        server.screen(viewer)?;
        server.input(viewer, json!({"kind":"paste","text":"stty size"}))?;
        server.enter(viewer)?;
        eventually(|| Ok(server.screen(viewer)?.contains("11 40")))?;
        if despawn {
            server.rpc("world.despawn_entity", json!({"entity":small}))?;
        } else {
            server.rpc(
                "world.remove_components",
                json!({"entity":small,"components":["fux::model::Viewer"]}),
            )?;
        }
        assert_eq!(
            server.rpc("fux.frame", json!({"viewer":small}))?,
            json!({"paint":"", "detach":true})
        );
        server.screen(viewer)?;
        server.input(viewer, json!({"kind":"paste","text":"clear; stty size"}))?;
        server.enter(viewer)?;
        eventually(|| Ok(server.screen(viewer)?.contains("23 80")))?;
    }
    Ok(())
}

#[test]
fn repeated_copy_effects_are_not_replaceable_paints() -> Outcome {
    use std::io::{BufRead, BufReader};
    let server = Server::start()?;
    let viewer = server.attach()?;
    let socket = server.socket.clone();
    let (sender, received) = std::sync::mpsc::channel();
    let reader = thread::spawn(move || -> Result<(), String> {
        let response = unix_http::agent(&socket, Some(Duration::from_secs(5)))
            .post(unix_http::URL)
            .send_json(json!({
                "jsonrpc":"2.0", "id":2, "method":"fux.frame+watch", "params":{"viewer":viewer}
            }))
            .need()?;
        for line in BufReader::new(response.into_body().into_reader()).lines() {
            let line = line.need()?;
            if let Some(data) = line.strip_prefix("data:") {
                let value: Value = serde_json::from_str(data).need()?;
                let paint = value.at("result").at("paint");
                let paint = paint.as_str().need()?;
                if sender.send(paint.matches("\x1b]52;").count()).is_err() {
                    break;
                }
            }
        }
        Ok(())
    });
    assert_eq!(received.recv_timeout(Duration::from_secs(5))?, 0);
    // These are explicit equal copies, not four interchangeable paint snapshots.
    for _ in 0..4 {
        server.command(viewer, "copy")?;
    }
    let mut copies = 0;
    while copies < 4 {
        copies += received.recv_timeout(Duration::from_secs(5))?;
    }
    assert_eq!(copies, 4);
    // Direct snapshots must not replay effects already delivered to the stream.
    assert!(
        !server
            .rpc("fux.frame", json!({"viewer":viewer}))?
            .at("paint")
            .as_str()
            .need()?
            .contains("\x1b]52;")
    );
    drop(received);
    let workspace = server.workspace_of(viewer)?;
    server.control(
        viewer,
        json!({"kind":"rename","subject":{"workspace":workspace},"name":"reader-finished"}),
    )?;
    reader.join().map_err(|_| "watch reader panicked")??;
    Ok(())
}

#[test]
fn blocked_terminal_paint_does_not_block_stream_drain() -> Outcome {
    use portable_pty::{Child as PtyChild, CommandBuilder, MasterPty, PtySize, native_pty_system};
    use std::io::Read;

    struct SlowTerminal {
        child: Box<dyn PtyChild + Send + Sync>,
        master: Option<Box<dyn MasterPty + Send>>,
    }
    impl Drop for SlowTerminal {
        fn drop(&mut self) {
            self.master.take();
            let _ = self.child.kill();
            let _ = self.child.wait();
        }
    }

    let server = Server::start()?;
    let pair = native_pty_system().openpty(PtySize {
        rows: 24,
        cols: 80,
        pixel_width: 0,
        pixel_height: 0,
    })?;
    let mut reader = pair.master.try_clone_reader()?;
    let mut command = CommandBuilder::new(env!("CARGO_BIN_EXE_fux"));
    command.arg("attach");
    command.env("FUX_SOCKET", &server.socket);
    command.env_remove("FUX_ENDPOINT");
    let mut terminal = SlowTerminal {
        child: {
            let _spawn = SPAWN.lock().unwrap_or_else(|error| error.into_inner());
            pair.slave.spawn_command(command)?
        },
        master: Some(pair.master),
    };
    drop(pair.slave);
    let (ready, received) = std::sync::mpsc::sync_channel(1);
    let first_paint = thread::spawn(move || {
        let mut output = Vec::new();
        let mut chunk = [0; 4096];
        while let Ok(count) = reader.read(&mut chunk) {
            if count == 0 {
                break;
            }
            output.extend_from_slice(chunk.get(..count).unwrap_or_default());
            if output.windows(8).any(|bytes| bytes == b"\x1b[?2026l") {
                let _ = ready.send(());
                return;
            }
        }
    });
    received.recv_timeout(Duration::from_secs(5))?;
    first_paint
        .join()
        .map_err(|_| "first paint reader panicked")?;
    let viewer = server
        .query("fux::model::Viewer")?
        .at(0)
        .at("entity")
        .as_u64()
        .need()?;
    let original = server
        .query("fux::model::ProcessState")?
        .at(0)
        .at("entity")
        .clone();
    server.split(viewer, "vertical", Some("exec /usr/bin/yes load"))?;
    // Deliberately stop reading the outer PTY. Paint will block, but HTTP must
    // continue draining/coalescing rather than fill Bevy's watch channel.
    thread::sleep(Duration::from_millis(500));
    assert!(
        server
            .query("fux::model::Viewer")?
            .rows()
            .any(|row| row.at("entity") == viewer)
    );
    assert!(terminal.child.try_wait()?.is_none());
    let states = server.query("fux::model::ProcessState")?;
    let hot = states
        .rows()
        .find(|row| row.at("entity") != original)
        .need()?;
    let entity = hot.at("entity");
    let pid = hot
        .at("components")
        .at("fux::model::ProcessState")
        .at("status")
        .at("pid")
        .as_u64()
        .need()? as i32;
    server.command(viewer, "terminate")?;
    eventually(|| Ok(!alive(pid)))?;
    let states = server.query("fux::model::ProcessState")?;
    let state = &states
        .rows()
        .find(|row| row.at("entity") == entity)
        .need()?
        .at("components")
        .at("fux::model::ProcessState");
    assert!(state.at("status").at("pid").is_null(), "{state}");
    assert_eq!(state.at("status").at("kind"), "exited", "{state}");
    assert!(state.at("status").at("code").is_number(), "{state}");
    Ok(())
}

/// The components on one entity, as non-strict `world.get_components` reports them.
fn components(server: &Server, entity: u64, names: &[&str]) -> Result<Value, String> {
    Ok(server
        .rpc(
            "world.get_components",
            json!({"entity":entity,"components":names}),
        )?
        .at("components"))
}

fn parent_of(server: &Server, entity: u64) -> Result<Option<u64>, String> {
    Ok(
        components(server, entity, &["bevy_ecs::hierarchy::ChildOf"])?
            .at("bevy_ecs::hierarchy::ChildOf")
            .as_u64(),
    )
}

/// The viewer names a workspace, a tab listed under it, and a pane view inside
/// that tab: membership, not just the presence of the three components.
fn assert_viewer_consistent(server: &Server, viewer: u64) -> Outcome {
    let workspace = server.workspace_of(viewer)?;
    let tab = server.on_tab(viewer)?.as_u64().need()?;
    let focus = server.focused(viewer)?.as_u64().need()?;
    assert!(
        server
            .query("fux::model::Workspace")?
            .rows()
            .any(|row| row.at("entity") == workspace)
    );
    assert!(
        server
            .query("fux::model::Tab")?
            .rows()
            .any(|row| row.at("entity") == tab)
    );
    assert_eq!(parent_of(server, tab)?, Some(workspace));
    assert!(
        components(server, focus, &["fux::model::PaneView"])?
            .get("fux::model::PaneView")
            .is_some()
    );
    let mut cursor = focus;
    let mut depth = 0;
    while cursor != tab {
        cursor = parent_of(server, cursor)?.ok_or("focused pane is not inside its tab")?;
        depth += 1;
        assert!(depth < 64, "focused pane is not inside its tab");
    }
    Ok(())
}

const VIEWER: &str = "fux::model::Viewer";
fn viewer_value() -> Value {
    json!({"rows":24,"cols":80,"zoom":false,"scrollback":0,"notice":null})
}

/// Hunt 5 finding 001: a `Viewer` on a layout entity, then any viewer
/// relationship inserted on a real viewer, overflowed the stack and aborted
/// the server. Each combination runs on a fresh server, bounded by `eventually`.
fn viewer_on_layout_entity(layout: &str) -> Outcome {
    for relation in ["Viewing", "OnTab", "Focused"] {
        for order in ["viewer_first", "layout_last", "both"] {
            let server = Server::start()?;
            let viewer = server.attach()?;
            server.split(viewer, "horizontal", None)?;
            eventually(|| Ok(server.query("fux::model::Split")?.rows().next().is_some()))?;
            let focus = server.focused(viewer)?.as_u64().need()?;
            let process = components(&server, focus, &["fux::model::PaneView"])?
                .at("fux::model::PaneView")
                .at("pane");
            let layout_value = match layout {
                "fux::model::PaneView" => json!({"pane":process}),
                _ => json!({}),
            };
            let node = match order {
                "viewer_first" => {
                    let node = match layout {
                        "fux::model::Workspace" => server.workspace_of(viewer)?,
                        "fux::model::Tab" => server.on_tab(viewer)?.as_u64().need()?,
                        "fux::model::PaneView" => focus,
                        _ => server
                            .query("fux::model::Split")?
                            .rows()
                            .next()
                            .need()?
                            .at("entity")
                            .as_u64()
                            .need()?,
                    };
                    server.rpc(
                        "world.insert_components",
                        json!({"entity":node,"components":{VIEWER:viewer_value()}}),
                    )?;
                    node
                }
                "layout_last" => {
                    let second = server.attach()?;
                    server.rpc(
                        "world.insert_components",
                        json!({"entity":second,"components":{layout:layout_value}}),
                    )?;
                    second
                }
                _ => server
                    .rpc(
                        "world.spawn_entity",
                        json!({"components":{layout:layout_value,VIEWER:viewer_value()}}),
                    )?
                    .at("entity")
                    .as_u64()
                    .need()?,
            };
            let value = match relation {
                "Viewing" => json!(server.workspace_of(viewer)?),
                "OnTab" => server.on_tab(viewer)?,
                _ => server.focused(viewer)?,
            };
            let component = format!("fux::model::{relation}");
            server.rpc(
                "world.insert_components",
                json!({"entity":viewer,"components":{component:value}}),
            )?;
            let context = format!("{layout} {relation} {order}");
            eventually(|| Ok(server.rpc("rpc.discover", Value::Null).is_ok()))
                .map_err(|e| format!("{context}: {e}"))?;
            let left = components(&server, node, &[VIEWER, layout])?;
            assert!(left.get(VIEWER).is_none(), "{context}: Viewer kept");
            assert!(left.get(layout).is_some(), "{context}: layout role lost");
            assert_viewer_consistent(&server, viewer).map_err(|e| format!("{context}: {e}"))?;
            assert!(!server.screen(viewer)?.is_empty(), "{context}");
        }
    }
    Ok(())
}

#[test]
fn a_viewer_on_a_tab_is_removed_and_the_server_survives() -> Outcome {
    viewer_on_layout_entity("fux::model::Tab")
}
#[test]
fn a_viewer_on_a_workspace_is_removed_and_the_server_survives() -> Outcome {
    viewer_on_layout_entity("fux::model::Workspace")
}
#[test]
fn a_viewer_on_a_pane_view_is_removed_and_the_server_survives() -> Outcome {
    viewer_on_layout_entity("fux::model::PaneView")
}
#[test]
fn a_viewer_on_a_split_is_removed_and_the_server_survives() -> Outcome {
    viewer_on_layout_entity("fux::model::Split")
}

/// Entity ids arrive as any u64. Bits no entity can have (0 among them) made
/// `Entity::from_bits` panic inside the ECS schedule, ending every session.
#[test]
fn frame_requests_for_impossible_entity_ids_are_refused() -> Outcome {
    let server = Server::start()?;
    let viewer = server.attach()?;
    for id in [0_u64, 0xFFFF_FFFF_0000_0000] {
        let error = server
            .rpc("fux.frame", json!({"viewer":id}))
            .err()
            .unwrap_or_default();
        assert!(error.contains("not an entity id"), "{id}: {error}");
        // The watch's error arrives as an event; what matters is that the
        // request bridge survives to carry the next request.
        let agent = unix_http::agent(&server.socket, Some(Duration::from_secs(1)));
        if let Ok(mut response) = agent.post(unix_http::URL).send_json(
            json!({"jsonrpc":"2.0","id":1,"method":"fux.frame+watch","params":{"viewer":id}}),
        ) {
            let _ = response.body_mut().read_to_string();
        }
        assert!(server.rpc("rpc.discover", Value::Null).is_ok());
    }
    assert!(!server.screen(viewer)?.is_empty());
    Ok(())
}