supercode-cli 0.4.6

//! Process proof for one resumed SDK runtime observed/driven through a
//! separately launched ACP bridge. Disconnecting ACP must leave the hosted
//! process alive until explicit SDK shutdown.

use std::fs;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
#[cfg(unix)]
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
#[cfg(unix)]
use std::os::unix::process::CommandExt;
use std::path::PathBuf;
use std::process::{Child, Command, ExitStatus, Output, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

fn bin() -> &'static str {
    env!("CARGO_BIN_EXE_supercode")
}

static TEMP_HOME_SEQUENCE: AtomicU64 = AtomicU64::new(0);

fn temp_home() -> PathBuf {
    let nonce = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_nanos();
    let sequence = TEMP_HOME_SEQUENCE.fetch_add(1, Ordering::Relaxed);
    let home = std::env::temp_dir().join(format!(
        "supercode-attachable-runtime-{}-{nonce}-{sequence}",
        std::process::id()
    ));
    fs::create_dir_all(&home).unwrap();
    home
}

#[cfg(unix)]
fn free_loopback_port() -> u16 {
    TcpListener::bind("127.0.0.1:0")
        .unwrap()
        .local_addr()
        .unwrap()
        .port()
}

#[cfg(unix)]
struct OwnedProcess {
    child: Child,
}

#[cfg(unix)]
impl OwnedProcess {
    fn new(child: Child) -> Self {
        Self { child }
    }
}

#[cfg(unix)]
impl Drop for OwnedProcess {
    fn drop(&mut self) {
        if self.child.try_wait().ok().flatten().is_none() {
            self.child.kill().ok();
        }
        self.child.wait().ok();
    }
}

#[cfg(unix)]
struct OpenSshFixture {
    root: PathBuf,
    user: String,
    client_key: PathBuf,
    daemon: OwnedProcess,
    port: u16,
}

#[cfg(unix)]
impl OpenSshFixture {
    fn start(root: PathBuf) -> Self {
        fs::create_dir_all(&root).unwrap();
        let host_key = root.join("host_key");
        let client_key = root.join("client_key");
        for key in [&host_key, &client_key] {
            let output = Command::new("ssh-keygen")
                .args(["-q", "-t", "ed25519", "-N", "", "-f"])
                .arg(key)
                .output()
                .unwrap();
            assert!(output.status.success(), "{:?}", output.stderr);
        }
        let authorized_keys = root.join("authorized_keys");
        fs::copy(client_key.with_extension("pub"), &authorized_keys).unwrap();
        let port = free_loopback_port();
        let user = std::env::var("USER").unwrap();
        let mut command = Command::new("/usr/sbin/sshd");
        command
            .args(["-D", "-e", "-f", "/dev/null", "-h"])
            .arg(&host_key)
            .args(["-p", &port.to_string()])
            .args(["-o", "ListenAddress=127.0.0.1"])
            .args([
                "-o",
                &format!("AuthorizedKeysFile={}", authorized_keys.display()),
            ])
            .args(["-o", "StrictModes=no"])
            .args(["-o", "PasswordAuthentication=no"])
            .args(["-o", "KbdInteractiveAuthentication=no"])
            .args(["-o", "PubkeyAuthentication=yes"])
            .args(["-o", "UsePAM=no"])
            .args([
                "-o",
                &format!("PidFile={}", root.join("sshd.pid").display()),
            ])
            .args(["-o", "LogLevel=ERROR"])
            .args(["-o", "PermitRootLogin=no"])
            .args(["-o", &format!("AllowUsers={user}")])
            .args(["-o", "AllowTcpForwarding=local"])
            .args(["-o", "GatewayPorts=no"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::inherit());
        let mut daemon = OwnedProcess::new(command.spawn().unwrap());
        let deadline = Instant::now() + Duration::from_secs(5);
        loop {
            if TcpStream::connect(("127.0.0.1", port)).is_ok() {
                break;
            }
            assert!(
                daemon.child.try_wait().unwrap().is_none(),
                "test sshd exited during startup"
            );
            assert!(Instant::now() < deadline, "test sshd startup timed out");
            std::thread::sleep(Duration::from_millis(10));
        }
        Self {
            root,
            user,
            client_key,
            daemon,
            port,
        }
    }

    fn tunnel(&self, destination: SocketAddr) -> (OwnedProcess, SocketAddr) {
        let local = SocketAddr::from(([127, 0, 0, 1], free_loopback_port()));
        let mut command = Command::new("ssh");
        command
            .args(["-N", "-T", "-p", &self.port.to_string(), "-i"])
            .arg(&self.client_key)
            .args(["-o", "BatchMode=yes"])
            .args(["-o", "StrictHostKeyChecking=no"])
            .args(["-o", "UserKnownHostsFile=/dev/null"])
            .args(["-o", "LogLevel=ERROR"])
            .args(["-o", "ExitOnForwardFailure=yes"])
            .args(["-L", &format!("{local}:{destination}")])
            .arg(format!("{}@127.0.0.1", self.user))
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::inherit());
        let mut tunnel = OwnedProcess::new(command.spawn().unwrap());
        let deadline = Instant::now() + Duration::from_secs(5);
        loop {
            if TcpStream::connect(local).is_ok() {
                break;
            }
            assert!(
                tunnel.child.try_wait().unwrap().is_none(),
                "operator-created SSH tunnel exited during startup"
            );
            assert!(Instant::now() < deadline, "SSH tunnel startup timed out");
            std::thread::sleep(Duration::from_millis(10));
        }
        (tunnel, local)
    }

    fn credential_stream(
        &self,
        path: &std::path::Path,
    ) -> (OwnedProcess, std::process::ChildStdout) {
        let mut command = Command::new("ssh");
        command
            .args(["-T", "-p", &self.port.to_string(), "-i"])
            .arg(&self.client_key)
            .args(["-o", "BatchMode=yes"])
            .args(["-o", "StrictHostKeyChecking=no"])
            .args(["-o", "UserKnownHostsFile=/dev/null"])
            .args(["-o", "LogLevel=ERROR"])
            .arg(format!("{}@127.0.0.1", self.user))
            .arg(format!("cat {}", path.display()))
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::null());
        let mut process = OwnedProcess::new(command.spawn().unwrap());
        let stdout = process.child.stdout.take().unwrap();
        (process, stdout)
    }
}

#[cfg(unix)]
impl Drop for OpenSshFixture {
    fn drop(&mut self) {
        let _ = &self.daemon;
        fs::remove_dir_all(&self.root).ok();
    }
}

fn capture_pipe<R: Read + Send + 'static>(reader: R) -> mpsc::Receiver<std::io::Result<Vec<u8>>> {
    let (sender, receiver) = mpsc::sync_channel(1);
    std::thread::spawn(move || {
        const MAX_CAPTURE_BYTES: u64 = 1024 * 1024;
        let mut bytes = Vec::new();
        let result = reader
            .take(MAX_CAPTURE_BYTES + 1)
            .read_to_end(&mut bytes)
            .and_then(|_| {
                if bytes.len() as u64 > MAX_CAPTURE_BYTES {
                    Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "subprocess output exceeded 1 MiB",
                    ))
                } else {
                    Ok(bytes)
                }
            });
        let _ = sender.send(result);
    });
    receiver
}

fn recv_capture(
    receiver: mpsc::Receiver<std::io::Result<Vec<u8>>>,
    stream: &str,
) -> std::io::Result<Vec<u8>> {
    receiver
        .recv_timeout(Duration::from_secs(2))
        .map_err(|error| {
            std::io::Error::new(
                std::io::ErrorKind::TimedOut,
                format!("{stream} capture did not finish: {error}"),
            )
        })?
}

fn output_bounded(command: &mut Command, timeout: Duration) -> std::io::Result<Output> {
    let label = command.get_program().to_string_lossy().into_owned();
    let mut child = command
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;
    let stdout = capture_pipe(child.stdout.take().unwrap());
    let stderr = capture_pipe(child.stderr.take().unwrap());
    let deadline = Instant::now() + timeout;
    let status = loop {
        if let Some(status) = child.try_wait()? {
            break status;
        }
        if Instant::now() >= deadline {
            child.kill().ok();
            child.wait().ok();
            return Err(std::io::Error::new(
                std::io::ErrorKind::TimedOut,
                format!("{label} did not exit within {timeout:?}"),
            ));
        }
        std::thread::sleep(Duration::from_millis(10));
    };
    Ok(Output {
        status,
        stdout: recv_capture(stdout, "stdout")?,
        stderr: recv_capture(stderr, "stderr")?,
    })
}

fn wait_child_bounded(child: &mut Child, timeout: Duration, context: &str) -> ExitStatus {
    let deadline = Instant::now() + timeout;
    loop {
        if let Some(status) = child.try_wait().unwrap() {
            return status;
        }
        if Instant::now() >= deadline {
            child.kill().ok();
            let kill_deadline = Instant::now() + Duration::from_secs(2);
            while Instant::now() < kill_deadline {
                if child.try_wait().ok().flatten().is_some() {
                    break;
                }
                std::thread::sleep(Duration::from_millis(10));
            }
            panic!("{context} did not exit within {timeout:?}");
        }
        std::thread::sleep(Duration::from_millis(10));
    }
}

fn join_thread_bounded(task: std::thread::JoinHandle<()>, timeout: Duration, context: &str) {
    let deadline = Instant::now() + timeout;
    while !task.is_finished() {
        assert!(
            Instant::now() < deadline,
            "{context} did not finish within {timeout:?}"
        );
        std::thread::sleep(Duration::from_millis(10));
    }
    task.join().unwrap();
}

#[cfg(unix)]
fn set_nonblocking(stream: &impl AsRawFd) {
    let fd = stream.as_raw_fd();
    // SAFETY: modifies only the supplied owned descriptor's status flags.
    unsafe {
        let flags = libc::fcntl(fd, libc::F_GETFL);
        assert!(flags >= 0);
        assert_eq!(libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK), 0);
    }
}

#[cfg(unix)]
#[derive(Clone, Debug, Eq, PartialEq)]
struct TermiosSnapshot {
    iflag: libc::tcflag_t,
    oflag: libc::tcflag_t,
    cflag: libc::tcflag_t,
    lflag: libc::tcflag_t,
    cc: [libc::cc_t; libc::NCCS],
    ispeed: libc::speed_t,
    ospeed: libc::speed_t,
}

#[cfg(unix)]
struct Pty {
    master: OwnedFd,
    slave: RawFd,
}

#[cfg(unix)]
impl Pty {
    fn open(width: u16, height: u16) -> Self {
        let mut master = -1;
        let mut slave = -1;
        let mut winsize = libc::winsize {
            ws_row: height,
            ws_col: width,
            ws_xpixel: 0,
            ws_ypixel: 0,
        };
        // SAFETY: valid out pointers and a fully initialized winsize.
        let result = unsafe {
            libc::openpty(
                &mut master,
                &mut slave,
                std::ptr::null_mut(),
                std::ptr::null_mut(),
                std::ptr::addr_of_mut!(winsize),
            )
        };
        assert_eq!(result, 0, "openpty: {}", std::io::Error::last_os_error());
        // Exercise the same explicit resize plumbing a real tmux pane uses.
        // SAFETY: master is a valid freshly opened PTY descriptor.
        assert_eq!(
            unsafe { libc::ioctl(master, libc::TIOCSWINSZ as _, &winsize) },
            0
        );
        Self {
            // SAFETY: ownership transfers from successful `openpty`.
            master: unsafe { OwnedFd::from_raw_fd(master) },
            slave,
        }
    }

    fn termios(&self) -> TermiosSnapshot {
        let mut value = std::mem::MaybeUninit::<libc::termios>::uninit();
        // SAFETY: master is valid and value is a correctly sized out pointer.
        assert_eq!(
            unsafe { libc::tcgetattr(self.master.as_raw_fd(), value.as_mut_ptr()) },
            0
        );
        // SAFETY: tcgetattr succeeded.
        let value = unsafe { value.assume_init() };
        TermiosSnapshot {
            iflag: value.c_iflag,
            oflag: value.c_oflag,
            cflag: value.c_cflag,
            lflag: value.c_lflag,
            cc: value.c_cc,
            // SAFETY: value is initialized termios state.
            ispeed: unsafe { libc::cfgetispeed(&value) },
            // SAFETY: value is initialized termios state.
            ospeed: unsafe { libc::cfgetospeed(&value) },
        }
    }

    fn write(&self, bytes: &[u8]) {
        let mut offset = 0;
        while offset < bytes.len() {
            // SAFETY: master and the remaining byte slice are valid.
            let count = unsafe {
                libc::write(
                    self.master.as_raw_fd(),
                    bytes[offset..].as_ptr().cast(),
                    bytes.len() - offset,
                )
            };
            assert!(count >= 0, "pty write: {}", std::io::Error::last_os_error());
            offset += count as usize;
        }
    }

    fn paste(&self, text: &str) {
        self.write(b"\x1b[200~");
        self.write(text.as_bytes());
        self.write(b"\x1b[201~");
    }

    fn paste_and_enter(&self, text: &str) {
        self.paste(text);
        std::thread::sleep(Duration::from_millis(20));
        self.write(b"\r");
    }

    fn expect(&self, output: &mut String, needle: &str) {
        self.expect_after(output, 0, needle);
    }

    fn expect_after(&self, output: &mut String, start: usize, needle: &str) {
        let fd = self.master.as_raw_fd();
        // SAFETY: modifies only this master fd's nonblocking status.
        unsafe {
            let flags = libc::fcntl(fd, libc::F_GETFL);
            libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK);
        }
        let deadline = std::time::Instant::now() + Duration::from_secs(15);
        while !output[start..].contains(needle) {
            assert!(
                std::time::Instant::now() < deadline,
                "timed out waiting for {needle:?}; output={output:?}"
            );
            if !self.drain(output) {
                std::thread::sleep(Duration::from_millis(10));
            }
        }
    }

    fn drain(&self, output: &mut String) -> bool {
        let mut buffer = [0_u8; 8192];
        // SAFETY: fd and output buffer are valid.
        let count = unsafe {
            libc::read(
                self.master.as_raw_fd(),
                buffer.as_mut_ptr().cast(),
                buffer.len(),
            )
        };
        if count <= 0 {
            return false;
        }
        let chunk = String::from_utf8_lossy(&buffer[..count as usize]);
        if chunk.contains("\x1b[6n") {
            // Crossterm's startup cursor query expects the terminal emulator
            // to answer, as a real tmux/xterm pane would.
            self.write(b"\x1b[1;1R");
        }
        output.push_str(&chunk);
        true
    }
}

#[cfg(unix)]
fn wait_success(child: &mut std::process::Child, pty: &Pty, output: &mut String) {
    let deadline = std::time::Instant::now() + Duration::from_secs(10);
    loop {
        if let Some(status) = child.try_wait().unwrap() {
            while pty.drain(output) {}
            assert!(status.success(), "{output}");
            return;
        }
        if std::time::Instant::now() >= deadline {
            child.kill().ok();
            child.wait().ok();
            while pty.drain(output) {}
            panic!("frontend did not exit after typed detach; output={output:?}");
        }
        pty.drain(output);
        std::thread::sleep(Duration::from_millis(10));
    }
}

#[cfg(unix)]
impl Drop for Pty {
    fn drop(&mut self) {
        // SAFETY: this object exclusively owns the raw slave descriptor.
        unsafe { libc::close(self.slave) };
    }
}

#[cfg(unix)]
fn spawn_pty_attach(
    pty: &Pty,
    home: &PathBuf,
    address: SocketAddr,
    token: &str,
) -> std::process::Child {
    spawn_pty_attach_via(pty, home, address, token, false)
}

#[cfg(unix)]
fn spawn_pty_acp_attach(
    pty: &Pty,
    home: &PathBuf,
    address: SocketAddr,
    token: &str,
) -> std::process::Child {
    spawn_pty_attach_via(pty, home, address, token, true)
}

#[cfg(unix)]
fn spawn_pty_attach_via(
    pty: &Pty,
    home: &PathBuf,
    address: SocketAddr,
    token: &str,
    through_acp: bool,
) -> std::process::Child {
    let stdio = |fd| {
        // SAFETY: dup creates a new owned descriptor.
        let duplicated = unsafe { libc::dup(fd) };
        assert!(duplicated >= 0);
        // SAFETY: duplicated is newly owned.
        Stdio::from(unsafe { OwnedFd::from_raw_fd(duplicated) })
    };
    let slave = pty.slave;
    let mut command = Command::new(bin());
    command
        .env("HOME", home)
        .env("SUPERCODE_HOME", home.join("supercode-home"));
    command.args(["attach", "--connect", &format!("http://{address}")]);
    if through_acp {
        command.arg("--acp");
    }
    command
        .args(["--token", token])
        .stdin(stdio(slave))
        .stdout(stdio(slave))
        .stderr(stdio(slave));
    // SAFETY: async-signal-safe session/controlling-terminal setup between
    // fork and exec; slave remains inherited and valid.
    unsafe {
        command.pre_exec(move || {
            if libc::setsid() == -1 {
                return Err(std::io::Error::last_os_error());
            }
            if libc::ioctl(libc::STDIN_FILENO, libc::TIOCSCTTY as _, 0) == -1 {
                return Err(std::io::Error::last_os_error());
            }
            Ok(())
        });
    }
    command.spawn().unwrap()
}

#[cfg(unix)]
fn spawn_pty_registry_attach(
    pty: &Pty,
    home: &PathBuf,
    supercode_home: &PathBuf,
    runtime_id: &str,
) -> std::process::Child {
    spawn_pty_ui_attach(
        pty,
        home,
        supercode_home,
        runtime_id,
        "embedded",
        "--take-control",
    )
}

#[cfg(unix)]
fn spawn_pty_ui_attach(
    pty: &Pty,
    home: &PathBuf,
    supercode_home: &PathBuf,
    runtime_id: &str,
    ui: &str,
    authority: &str,
) -> std::process::Child {
    let stdio = |fd| {
        // SAFETY: dup creates a new owned descriptor.
        let duplicated = unsafe { libc::dup(fd) };
        assert!(duplicated >= 0);
        // SAFETY: duplicated is newly owned.
        Stdio::from(unsafe { OwnedFd::from_raw_fd(duplicated) })
    };
    let slave = pty.slave;
    let mut command = Command::new(bin());
    command
        .env("HOME", home)
        .env("SUPERCODE_HOME", supercode_home)
        .env("TERM", "xterm-256color")
        .args(["attach", runtime_id, "--ui", ui, authority])
        .stdin(stdio(slave))
        .stdout(stdio(slave))
        .stderr(stdio(slave));
    // SAFETY: async-signal-safe session/controlling-terminal setup between
    // fork and exec; slave remains inherited and valid.
    unsafe {
        command.pre_exec(move || {
            if libc::setsid() == -1 {
                return Err(std::io::Error::last_os_error());
            }
            if libc::ioctl(libc::STDIN_FILENO, libc::TIOCSCTTY as _, 0) == -1 {
                return Err(std::io::Error::last_os_error());
            }
            Ok(())
        });
    }
    command.spawn().unwrap_or_else(|error| {
        let fd_state = unsafe { libc::fcntl(slave, libc::F_GETFD) };
        panic!("spawning {ui} {authority} frontend on slave fd {slave} (state {fd_state}): {error}")
    })
}

#[cfg(unix)]
struct TmuxServer(String);

#[cfg(unix)]
impl TmuxServer {
    fn command(&self, args: &[&str]) -> std::process::Output {
        let mut command = Command::new("tmux");
        command.args(["-L", &self.0]).args(args);
        output_bounded(&mut command, Duration::from_secs(3)).unwrap()
    }

    fn expect(&self, needle: &str) -> String {
        self.expect_count(needle, 1)
    }

    fn expect_count(&self, needle: &str, count: usize) -> String {
        let deadline = std::time::Instant::now() + Duration::from_secs(15);
        loop {
            let output = self.command(&["capture-pane", "-p", "-S", "-100", "-t", "frontend"]);
            let rendered = String::from_utf8_lossy(&output.stdout).into_owned();
            if output.status.success() && rendered.matches(needle).count() >= count {
                return rendered;
            }
            assert!(
                std::time::Instant::now() < deadline,
                "timed out waiting for {count} copies of {needle:?} in tmux; output={rendered:?}; stderr={:?}",
                String::from_utf8_lossy(&output.stderr)
            );
            std::thread::sleep(Duration::from_millis(20));
        }
    }

    fn expect_after(&self, before: &str, after: &str) -> String {
        let deadline = std::time::Instant::now() + Duration::from_secs(15);
        loop {
            let output = self.command(&["capture-pane", "-p", "-S", "-100", "-t", "frontend"]);
            let rendered = String::from_utf8_lossy(&output.stdout).into_owned();
            let ordered = rendered
                .find(before)
                .is_some_and(|position| rendered[position..].contains(after));
            if output.status.success() && ordered {
                return rendered;
            }
            assert!(
                std::time::Instant::now() < deadline,
                "timed out waiting for {after:?} after {before:?} in tmux; output={rendered:?}; stderr={:?}",
                String::from_utf8_lossy(&output.stderr)
            );
            std::thread::sleep(Duration::from_millis(20));
        }
    }

    fn send_line(&self, line: &str) {
        let pasted = format!("\u{1b}[200~{line}\u{1b}[201~");
        let literal = self.command(&["send-keys", "-t", "frontend", "-l", &pasted]);
        assert!(literal.status.success(), "{:?}", literal.stderr);
        std::thread::sleep(Duration::from_millis(50));
        let enter = self.command(&["send-keys", "-t", "frontend", "Enter"]);
        assert!(enter.status.success(), "{:?}", enter.stderr);
    }

    fn send_tui_line(&self, line: &str) {
        let literal = self.command(&["send-keys", "-t", "frontend", "-l", line]);
        assert!(literal.status.success(), "{:?}", literal.stderr);
        std::thread::sleep(Duration::from_millis(50));
        let enter = self.command(&["send-keys", "-t", "frontend", "Enter"]);
        assert!(enter.status.success(), "{:?}", enter.stderr);
    }

    fn clear_screen_and_history(&self) {
        let clear_screen = self.command(&["send-keys", "-t", "frontend", "C-l"]);
        assert!(clear_screen.status.success(), "{:?}", clear_screen.stderr);
        let clear_history = self.command(&["clear-history", "-t", "frontend"]);
        assert!(clear_history.status.success(), "{:?}", clear_history.stderr);
    }

    fn wait_for_exit(&self) {
        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        loop {
            if !self
                .command(&["has-session", "-t", "frontend"])
                .status
                .success()
            {
                return;
            }
            assert!(
                std::time::Instant::now() < deadline,
                "tmux frontend did not exit after typed detach"
            );
            std::thread::sleep(Duration::from_millis(20));
        }
    }

    fn pane_pid(&self) -> u32 {
        let output = self.command(&["display-message", "-p", "-t", "frontend", "#{pane_pid}"]);
        assert!(output.status.success(), "{:?}", output.stderr);
        String::from_utf8(output.stdout)
            .unwrap()
            .trim()
            .parse()
            .unwrap()
    }

    fn termios(&self) -> TermiosSnapshot {
        let output = self.command(&["display-message", "-p", "-t", "frontend", "#{pane_tty}"]);
        assert!(output.status.success(), "{:?}", output.stderr);
        let tty = String::from_utf8(output.stdout).unwrap();
        let tty = tty.trim();
        let file = fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(tty)
            .unwrap();
        let fd = file.as_raw_fd();
        let mut value = std::mem::MaybeUninit::<libc::termios>::uninit();
        assert_eq!(unsafe { libc::tcgetattr(fd, value.as_mut_ptr()) }, 0);
        let value = unsafe { value.assume_init() };
        TermiosSnapshot {
            iflag: value.c_iflag,
            oflag: value.c_oflag,
            cflag: value.c_cflag,
            lflag: value.c_lflag,
            cc: value.c_cc,
            ispeed: unsafe { libc::cfgetispeed(&value) },
            ospeed: unsafe { libc::cfgetospeed(&value) },
        }
    }
}

#[cfg(unix)]
fn shell_word(value: &str) -> String {
    format!("'{}'", value.replace('\'', "'\"'\"'"))
}

#[cfg(unix)]
fn only_child_pid(parent: u32) -> libc::pid_t {
    let output = Command::new("pgrep")
        .args(["-P", &parent.to_string()])
        .output()
        .unwrap();
    assert!(output.status.success(), "no child for pid {parent}");
    let children = String::from_utf8(output.stdout).unwrap();
    let mut children = children.lines();
    let child = children.next().unwrap().parse().unwrap();
    assert!(children.next().is_none(), "multiple children for {parent}");
    child
}

#[cfg(unix)]
impl Drop for TmuxServer {
    fn drop(&mut self) {
        let mut command = Command::new("tmux");
        command.args(["-L", &self.0, "kill-server"]);
        let _ = output_bounded(&mut command, Duration::from_secs(3));
    }
}

fn accept_bounded(listener: &TcpListener, timeout: Duration, round: usize) -> TcpStream {
    let deadline = Instant::now() + timeout;
    loop {
        match listener.accept() {
            Ok((socket, _)) => return socket,
            Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                assert!(
                    Instant::now() < deadline,
                    "provider was not contacted for round {round} within {timeout:?}"
                );
                std::thread::sleep(Duration::from_millis(10));
            }
            Err(error) => panic!("provider accept failed in round {round}: {error}"),
        }
    }
}

fn spawn_provider(
    tmux_available: bool,
) -> (
    SocketAddr,
    std::thread::JoinHandle<()>,
    mpsc::Receiver<&'static str>,
    mpsc::Sender<()>,
) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    listener.set_nonblocking(true).unwrap();
    let address = listener.local_addr().unwrap();
    let (interrupt_started_tx, interrupt_started_rx) = mpsc::channel();
    let (release_first_turn_tx, release_first_turn_rx) = mpsc::channel();
    let task = std::thread::spawn(move || {
        let error_round = if tmux_available { 15 } else { 10 };
        for round in 0..=error_round {
            let mut socket = accept_bounded(&listener, Duration::from_secs(20), round);
            socket.set_nonblocking(false).unwrap();
            socket
                .set_read_timeout(Some(Duration::from_secs(10)))
                .unwrap();
            socket
                .set_write_timeout(Some(Duration::from_secs(10)))
                .unwrap();
            let request = read_http_request(&mut socket);
            let request = String::from_utf8_lossy(&request);
            if round == error_round {
                assert!(request.contains("ERROR_PROMPT"), "{request}");
                // Dropping a live provider connection produces a real
                // turn_failed lifecycle event without stopping the runtime.
                continue;
            }
            if round == 0 {
                assert!(request.contains("ATTACHED_ACP_PROMPT"), "{request}");
                let first = "data: {\"choices\":[{\"delta\":{\"content\":\"SHARED \"}}]}\n\n";
                let rest = "data: {\"choices\":[{\"delta\":{\"content\":\"RUNTIME DONE\"}}]}\n\ndata: [DONE]\n\n";
                let _ = write!(
                    socket,
                    "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{first}",
                    first.len() + rest.len()
                );
                socket.flush().unwrap();
                // Leave the turn genuinely in flight while the first ACP
                // client acknowledges and disconnects after this chunk.
                release_first_turn_rx
                    .recv_timeout(Duration::from_secs(15))
                    .expect("replacement ACP frontend did not attach before provider release");
                socket.write_all(rest.as_bytes()).unwrap();
                socket.flush().unwrap();
                continue;
            }
            if round == 9 {
                assert!(request.contains("LINE_ERROR_PROMPT"), "{request}");
                // Give the redirected line frontend its own real failure;
                // the later ERROR_PROMPT belongs to the full-screen path.
                continue;
            }
            let sse = match round {
                1 => {
                    assert!(request.contains("ACP_FRONTEND_TOOL_PROMPT"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_acp_frontend\",\"function\":{\"name\":\"list_dir\",\"arguments\":\"{}\"}}]}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                2 => {
                    assert!(request.contains("call_acp_frontend"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"content\":\"ACP FRONTEND DONE\"}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                3 => {
                    assert!(request.contains("ATTACHED_TERMINAL_PROMPT"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_pty\",\"function\":{\"name\":\"list_dir\",\"arguments\":\"{}\"}}]}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                4 => {
                    assert!(request.contains("call_pty"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"content\":\"LONG RESPONSE START — this response is intentionally long enough to wrap across several terminal rows and prove the shared viewport follows the live answer tail without hiding completion — TERMINAL RUNTIME DONE\"}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                5 => {
                    assert!(request.contains("INTERRUPT_PROMPT"), "{request}");
                    interrupt_started_tx.send("pty").unwrap();
                    std::thread::sleep(Duration::from_secs(2));
                    "data: {\"choices\":[{\"delta\":{\"content\":\"TOO LATE\"}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                6 => {
                    assert!(request.contains("LINE_TOOL_PROMPT"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_line\",\"function\":{\"name\":\"list_dir\",\"arguments\":\"{}\"}}]}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                7 => {
                    assert!(request.contains("call_line"), "{request}");
                    // The redirected human frontend must retain semantic
                    // content and ordering while stripping terminal title
                    // controls. `usage` is intentionally an unknown line
                    // event, so its adversarial extra payload must collapse
                    // to a bounded marker rather than raw JSON.
                    "data: {\"choices\":[{\"delta\":{\"content\":\"\\u001b]2;PWNED\\u0007LINE_SAFE\\nLINE_TOOL_DONE\"}}]}\n\ndata: {\"choices\":[{\"delta\":{}}],\"usage\":{\"prompt_tokens\":11,\"completion_tokens\":2,\"total_tokens\":13,\"adversarial\":{\"secret\":\"\\u001b]2;PAYLOAD\\u0007\"}}}\n\ndata: [DONE]\n\n".to_string()
                }
                8 => {
                    assert!(request.contains("LINE_INTERRUPT_PROMPT"), "{request}");
                    interrupt_started_tx.send("line").unwrap();
                    std::thread::sleep(Duration::from_secs(2));
                    "data: {\"choices\":[{\"delta\":{\"content\":\"LINE INTERRUPT TOO LATE\"}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                10 => {
                    assert!(request.contains("TMUX_PROMPT"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_tmux\",\"function\":{\"name\":\"list_dir\",\"arguments\":\"{}\"}}]}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                11 => {
                    assert!(request.contains("call_tmux"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"content\":\"TMUX RUNTIME DONE\"}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                12 => {
                    assert!(request.contains("TMUX_INTERRUPT_PROMPT"), "{request}");
                    interrupt_started_tx.send("tmux").unwrap();
                    std::thread::sleep(Duration::from_secs(2));
                    "data: {\"choices\":[{\"delta\":{\"content\":\"TMUX INTERRUPT TOO LATE\"}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                13 => {
                    assert!(request.contains("TMUX_APPROVAL_PROMPT"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_tmux_approval\",\"function\":{\"name\":\"bash\",\"arguments\":\"{\\\"command\\\":\\\"printf TMUX_APPROVED\\\"}\"}}]}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                14 => {
                    assert!(request.contains("TMUX_APPROVED"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"content\":\"TMUX APPROVAL DONE\"}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                _ => unreachable!(),
            };
            let _ = write!(
                socket,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{sse}",
                sse.len()
            );
        }
    });
    (address, task, interrupt_started_rx, release_first_turn_tx)
}

fn read_http_request(socket: &mut TcpStream) -> Vec<u8> {
    let deadline = Instant::now() + Duration::from_secs(10);
    let mut request = Vec::new();
    let mut buffer = [0_u8; 4096];
    loop {
        let remaining = deadline.saturating_duration_since(Instant::now());
        assert!(!remaining.is_zero(), "provider request read timed out");
        socket.set_read_timeout(Some(remaining)).unwrap();
        let count = socket
            .read(&mut buffer)
            .unwrap_or_else(|error| panic!("provider request read failed: {error}"));
        if count == 0 {
            break;
        }
        request.extend_from_slice(&buffer[..count]);
        assert!(
            request.len() <= 1024 * 1024,
            "provider request exceeded 1 MiB"
        );
        let Some(headers_end) = request.windows(4).position(|window| window == b"\r\n\r\n") else {
            continue;
        };
        let headers = String::from_utf8_lossy(&request[..headers_end]);
        let length = headers
            .lines()
            .find_map(|line| {
                let (name, value) = line.split_once(':')?;
                name.eq_ignore_ascii_case("content-length")
                    .then(|| value.trim().parse::<usize>().ok())
                    .flatten()
            })
            .unwrap_or(0);
        if request.len() >= headers_end + 4 + length {
            break;
        }
    }
    request
}

fn send(writer: &mut impl Write, value: serde_json::Value) {
    writeln!(writer, "{value}").unwrap();
    writer.flush().unwrap();
}

fn receive(reader: &mut impl BufRead) -> serde_json::Value {
    let deadline = Instant::now() + Duration::from_secs(15);
    let mut line = String::new();
    loop {
        match reader.read_line(&mut line) {
            Ok(0) => panic!("ACP bridge closed before its response"),
            Ok(_) => break,
            Err(error)
                if error.kind() == std::io::ErrorKind::WouldBlock
                    || error.kind() == std::io::ErrorKind::Interrupted =>
            {
                assert!(
                    Instant::now() < deadline,
                    "timed out waiting for ACP response"
                );
                std::thread::sleep(Duration::from_millis(10));
            }
            Err(error) => panic!("ACP bridge read failed: {error}"),
        }
    }
    serde_json::from_str(&line).unwrap()
}

fn read_line_until(reader: &mut impl BufRead, output: &mut String, needle: &str) {
    read_line_until_after(reader, output, 0, needle);
}

fn read_line_until_after(
    reader: &mut impl BufRead,
    output: &mut String,
    start: usize,
    needle: &str,
) {
    let deadline = std::time::Instant::now() + Duration::from_secs(15);
    while !output[start..].contains(needle) {
        assert!(
            std::time::Instant::now() < deadline,
            "timed out waiting for {needle:?}; output={output:?}"
        );
        let mut rendered_line = String::new();
        match reader.read_line(&mut rendered_line) {
            Ok(0) => panic!("line frontend closed before {needle:?}; output={output:?}"),
            Ok(_) => output.push_str(&rendered_line),
            Err(error)
                if error.kind() == std::io::ErrorKind::WouldBlock
                    || error.kind() == std::io::ErrorKind::Interrupted =>
            {
                // A nonblocking `read_line` may consume a valid partial line
                // before reporting `WouldBlock`. Preserve those bytes across
                // polls; dropping this local buffer loses streamed deltas.
                output.push_str(&rendered_line);
                std::thread::sleep(Duration::from_millis(10));
            }
            Err(error) => panic!("line frontend read failed: {error}; output={output:?}"),
        }
    }
}

fn read_to_string_bounded(
    reader: &mut impl Read,
    output: &mut String,
    timeout: Duration,
    context: &str,
) {
    const MAX_REMAINDER_BYTES: usize = 64 * 1024;
    let deadline = Instant::now() + timeout;
    let mut remainder = Vec::new();
    let mut buffer = [0_u8; 8192];
    loop {
        assert!(
            Instant::now() < deadline,
            "{context} did not reach EOF within {timeout:?}"
        );
        match reader.read(&mut buffer) {
            Ok(0) => break,
            Ok(count) => {
                remainder.extend_from_slice(&buffer[..count]);
                assert!(
                    remainder.len() <= MAX_REMAINDER_BYTES,
                    "{context} exceeded {MAX_REMAINDER_BYTES} bytes"
                );
            }
            Err(error)
                if error.kind() == std::io::ErrorKind::WouldBlock
                    || error.kind() == std::io::ErrorKind::Interrupted =>
            {
                std::thread::sleep(Duration::from_millis(10));
            }
            Err(error) => panic!("{context} read failed: {error}"),
        }
    }
    output.push_str(&String::from_utf8(remainder).unwrap());
}

fn http_rpc(
    address: SocketAddr,
    token: &str,
    id: u64,
    method: &str,
    params: serde_json::Value,
) -> serde_json::Value {
    http_rpc_as(address, token, None, id, method, params)
}

fn http_rpc_as(
    address: SocketAddr,
    token: &str,
    client_id: Option<&str>,
    id: u64,
    method: &str,
    params: serde_json::Value,
) -> serde_json::Value {
    let timeout = Duration::from_secs(2);
    let mut socket = TcpStream::connect_timeout(&address, timeout).unwrap();
    socket.set_read_timeout(Some(timeout)).unwrap();
    socket.set_write_timeout(Some(timeout)).unwrap();
    let body = serde_json::json!({"id": id, "method": method, "params": params}).to_string();
    let client_header = client_id
        .map(|client_id| format!("X-Supercode-Client-Id: {client_id}\r\n"))
        .unwrap_or_default();
    write!(
        socket,
        "POST /rpc HTTP/1.1\r\nHost: {address}\r\nAuthorization: Bearer {token}\r\n{client_header}Content-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    )
    .unwrap();
    const MAX_RPC_BYTES: usize = 1024 * 1024;
    let deadline = Instant::now() + timeout;
    let mut response = Vec::new();
    let mut buffer = [0_u8; 8192];
    loop {
        let remaining = deadline.saturating_duration_since(Instant::now());
        assert!(!remaining.is_zero(), "RPC response exceeded {timeout:?}");
        socket.set_read_timeout(Some(remaining)).unwrap();
        let count = socket
            .read(&mut buffer)
            .unwrap_or_else(|error| panic!("RPC response read failed: {error}"));
        if count == 0 {
            break;
        }
        response.extend_from_slice(&buffer[..count]);
        assert!(
            response.len() <= MAX_RPC_BYTES,
            "RPC response exceeded {MAX_RPC_BYTES} bytes"
        );
    }
    let response = String::from_utf8(response).unwrap();
    let body = response.split_once("\r\n\r\n").unwrap().1;
    serde_json::from_str(body).unwrap()
}

fn unauthorized_http_status(address: SocketAddr) -> String {
    let mut socket = TcpStream::connect_timeout(&address, Duration::from_secs(2)).unwrap();
    socket
        .set_read_timeout(Some(Duration::from_secs(2)))
        .unwrap();
    let body = serde_json::json!({"id":1,"method":"describe","params":{}}).to_string();
    write!(
        socket,
        "POST /rpc HTTP/1.1\r\nHost: {address}\r\nAuthorization: Bearer invalid\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    )
    .unwrap();
    let mut response = String::new();
    socket.read_to_string(&mut response).unwrap();
    response.lines().next().unwrap_or_default().to_string()
}

fn shutdown(address: SocketAddr, token: &str) {
    let response = http_rpc(address, token, 99, "shutdown", serde_json::json!({}));
    assert_eq!(response["result"]["shutting_down"], true);
}

fn spawn_echo_provider(request_count: usize) -> (SocketAddr, std::thread::JoinHandle<()>) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    listener.set_nonblocking(true).unwrap();
    let address = listener.local_addr().unwrap();
    let task = std::thread::spawn(move || {
        for round in 0..request_count {
            let mut socket = accept_bounded(&listener, Duration::from_secs(20), round);
            socket.set_nonblocking(false).unwrap();
            let request = read_http_request(&mut socket);
            assert!(
                String::from_utf8_lossy(&request).contains("REGISTRY_"),
                "{}",
                String::from_utf8_lossy(&request)
            );
            let sse = format!(
                "data: {{\"choices\":[{{\"delta\":{{\"content\":\"REGISTRY REPLY {round}\"}}}}]}}\n\ndata: [DONE]\n\n"
            );
            write!(
                socket,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{sse}",
                sse.len()
            )
            .unwrap();
        }
    });
    (address, task)
}

#[cfg(unix)]
fn spawn_dumb_terminal_provider() -> (SocketAddr, std::thread::JoinHandle<()>, mpsc::Receiver<()>) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    listener.set_nonblocking(true).unwrap();
    let address = listener.local_addr().unwrap();
    let (interrupt_started_tx, interrupt_started_rx) = mpsc::channel();
    let task = std::thread::spawn(move || {
        for round in 0..2 {
            let mut socket = accept_bounded(&listener, Duration::from_secs(20), round);
            socket
                .set_read_timeout(Some(Duration::from_secs(10)))
                .unwrap();
            socket
                .set_write_timeout(Some(Duration::from_secs(10)))
                .unwrap();
            let request = String::from_utf8_lossy(&read_http_request(&mut socket)).into_owned();
            let sse = if round == 0 {
                assert!(request.contains("DUMB_SUBMIT_PROMPT"), "{request}");
                "data: {\"choices\":[{\"delta\":{\"content\":\"DUMB SUBMIT DONE\"}}]}\n\ndata: [DONE]\n\n"
            } else {
                assert!(request.contains("DUMB_INTERRUPT_PROMPT"), "{request}");
                interrupt_started_tx.send(()).unwrap();
                std::thread::sleep(Duration::from_secs(2));
                "data: {\"choices\":[{\"delta\":{\"content\":\"DUMB INTERRUPT TOO LATE\"}}]}\n\ndata: [DONE]\n\n"
            };
            let _ = write!(
                socket,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{sse}",
                sse.len()
            );
        }
    });
    (address, task, interrupt_started_rx)
}

#[cfg(unix)]
fn write_private_credential(path: &std::path::Path, token: &str) {
    fs::write(path, token).unwrap();
    fs::set_permissions(path, fs::Permissions::from_mode(0o600)).unwrap();
}

fn spawn_goose_proof_provider() -> (SocketAddr, std::thread::JoinHandle<()>) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    listener.set_nonblocking(true).unwrap();
    let address = listener.local_addr().unwrap();
    let task = std::thread::spawn(move || {
        for round in 0..4 {
            let mut socket = accept_bounded(&listener, Duration::from_secs(30), round);
            socket.set_nonblocking(false).unwrap();
            let request = read_http_request(&mut socket);
            let request = String::from_utf8_lossy(&request);
            let sse = match round {
                0 => {
                    assert!(request.contains("GOOSE_PERMISSION_PROMPT"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"goose_proof_call\",\"function\":{\"name\":\"bash\",\"arguments\":\"{\\\"command\\\":\\\"printf GOOSE_TOOL_OK\\\"}\"}}]}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                1 => {
                    assert!(request.contains("goose_proof_call"), "{request}");
                    assert!(request.contains("GOOSE_TOOL_OK"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"content\":\"GOOSE_FIRST_DONE\"}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                2 => {
                    assert!(request.contains("GOOSE_RECONNECT_PROMPT"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"content\":\"GOOSE_RECONNECT_DONE\"}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                3 => {
                    assert!(request.contains("EMBEDDED_REPLACEMENT_PROMPT"), "{request}");
                    "data: {\"choices\":[{\"delta\":{\"content\":\"EMBEDDED_REPLACEMENT_DONE\"}}]}\n\ndata: [DONE]\n\n".to_string()
                }
                _ => unreachable!(),
            };
            write!(
                socket,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{sse}",
                sse.len()
            )
            .unwrap();
        }
    });
    (address, task)
}

fn spawn_registry_host(
    home: &std::path::Path,
    supercode_home: &std::path::Path,
    source: &std::path::Path,
    provider: SocketAddr,
    token: &str,
) -> (Child, SocketAddr, BufReader<std::process::ChildStderr>) {
    spawn_registry_host_with_ttl(home, supercode_home, source, provider, token, 100)
}

fn spawn_registry_host_with_ttl(
    home: &std::path::Path,
    supercode_home: &std::path::Path,
    source: &std::path::Path,
    provider: SocketAddr,
    token: &str,
    lease_ttl_ms: u64,
) -> (Child, SocketAddr, BufReader<std::process::ChildStderr>) {
    spawn_registry_host_with_ttl_and_iterations(
        home,
        supercode_home,
        source,
        provider,
        token,
        lease_ttl_ms,
        1,
    )
}

fn spawn_registry_host_with_ttl_and_iterations(
    home: &std::path::Path,
    supercode_home: &std::path::Path,
    source: &std::path::Path,
    provider: SocketAddr,
    token: &str,
    lease_ttl_ms: u64,
    max_iterations: usize,
) -> (Child, SocketAddr, BufReader<std::process::ChildStderr>) {
    let max_iterations = max_iterations.to_string();
    let mut child = Command::new(bin())
        .env("HOME", home)
        .env("SUPERCODE_HOME", supercode_home)
        .env_remove("OPENROUTER_API_KEY")
        .args([
            "--quiet",
            "--no-reduced",
            "--api-key",
            "x",
            "--base-url",
            &format!("http://{provider}"),
            "--max-iterations",
            &max_iterations,
            "resume",
            source.to_str().unwrap(),
            "--paused",
            "--serve",
            "--no-tmux",
            "--bind",
            "127.0.0.1:0",
            "--token",
            token,
            "--lease-ttl-ms",
            &lease_ttl_ms.to_string(),
        ])
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let stderr = child.stderr.take().unwrap();
    set_nonblocking(&stderr);
    let mut stderr = BufReader::new(stderr);
    let deadline = Instant::now() + Duration::from_secs(30);
    let address = loop {
        assert!(Instant::now() < deadline, "registry host startup timed out");
        let mut line = String::new();
        match stderr.read_line(&mut line) {
            Ok(0) => panic!("registry host exited before publishing its address"),
            Ok(_) => {
                if let Some(value) = line.split("listening on http://").nth(1) {
                    break value.trim().parse().unwrap();
                }
            }
            Err(error)
                if matches!(
                    error.kind(),
                    std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted
                ) =>
            {
                std::thread::sleep(Duration::from_millis(10));
            }
            Err(error) => panic!("registry host stderr failed: {error}"),
        }
    };
    (child, address, stderr)
}

fn open_frontend_events(address: SocketAddr, token: &str, client_id: &str) -> TcpStream {
    let timeout = Duration::from_secs(2);
    let mut socket = TcpStream::connect_timeout(&address, timeout).unwrap();
    socket.set_read_timeout(Some(timeout)).unwrap();
    socket.set_write_timeout(Some(timeout)).unwrap();
    write!(
        socket,
        "GET /frontend/events HTTP/1.1\r\nHost: {address}\r\nAuthorization: Bearer {token}\r\nX-Supercode-Client-Id: {client_id}\r\nConnection: close\r\n\r\n"
    )
    .unwrap();
    let mut header = Vec::new();
    let mut byte = [0_u8; 1];
    while !header.ends_with(b"\r\n\r\n") {
        socket.read_exact(&mut byte).unwrap();
        header.push(byte[0]);
        assert!(header.len() < 16 * 1024);
    }
    assert!(String::from_utf8(header)
        .unwrap()
        .starts_with("HTTP/1.1 200"));
    socket
}

fn read_frontend_events_until(
    socket: &mut TcpStream,
    expected_terminal_events: usize,
) -> Vec<serde_json::Value> {
    socket
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let mut reader = BufReader::new(socket);
    let mut events = Vec::new();
    let mut terminal_events = 0;
    while terminal_events < expected_terminal_events {
        let mut line = String::new();
        assert!(reader.read_line(&mut line).unwrap() > 0);
        let Some(payload) = line.strip_prefix("data: ") else {
            continue;
        };
        let event: serde_json::Value = serde_json::from_str(payload.trim()).unwrap();
        if matches!(
            event["kind"].as_str(),
            Some("turn_succeeded" | "turn_interrupted" | "turn_failed")
        ) {
            terminal_events += 1;
        }
        events.push(event);
    }
    events
}

fn list_live_sessions(
    home: &std::path::Path,
    supercode_home: &std::path::Path,
) -> Vec<serde_json::Value> {
    let mut command = Command::new(bin());
    command
        .env("HOME", home)
        .env("SUPERCODE_HOME", supercode_home)
        .args(["sessions", "list", "--live", "--json"]);
    let output = output_bounded(&mut command, Duration::from_secs(10)).unwrap();
    assert!(output.status.success(), "{:?}", output.stderr);
    serde_json::from_slice(&output.stdout).unwrap()
}

#[cfg(unix)]
struct TunneledCredential<'a> {
    ssh: &'a OpenSshFixture,
    path: &'a std::path::Path,
    token: &'a str,
    client_id: &'a str,
}

#[cfg(unix)]
fn run_tunneled_frontend(
    home: &std::path::Path,
    supercode_home: &std::path::Path,
    address: SocketAddr,
    credential: TunneledCredential<'_>,
    prompt: &str,
    expected_reply: &str,
    after_sequence: Option<u64>,
) -> Output {
    let (_credential_process, credential_stream) =
        credential.ssh.credential_stream(credential.path);
    let inherited_fd = 9;
    let mut command = Command::new(bin());
    command
        .env("HOME", home)
        .env("SUPERCODE_HOME", supercode_home)
        .env_remove("SUPERCODE_SERVER_TOKEN")
        .args([
            "--quiet",
            "attach",
            "--connect",
            &format!("http://{address}"),
            "--credential-fd",
            &inherited_fd.to_string(),
            "--client-id",
            credential.client_id,
        ]);
    if let Some(sequence) = after_sequence {
        command
            .arg("--acp")
            .args(["--after-sequence", &sequence.to_string()]);
    }
    let visible_args = command
        .get_args()
        .map(|argument| argument.to_string_lossy())
        .collect::<Vec<_>>();
    assert!(
        visible_args
            .iter()
            .all(|argument| !argument.contains(credential.token)),
        "bearer secret appeared in frontend argv: {visible_args:?}"
    );
    let source_fd = credential_stream.as_raw_fd();
    // SAFETY: the SSH stdout remains owned until spawn completes. dup2 installs a
    // non-CLOEXEC descriptor for the top-level attach and its ACP child.
    unsafe {
        command.pre_exec(move || {
            if libc::dup2(source_fd, inherited_fd) == -1 {
                return Err(std::io::Error::last_os_error());
            }
            Ok(())
        });
    }
    let mut child = command
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let process = Command::new("ps")
        .args(["eww", "-p", &child.id().to_string()])
        .output()
        .unwrap();
    assert!(
        !String::from_utf8_lossy(&process.stdout).contains(credential.token),
        "bearer secret appeared in frontend argv/environment"
    );
    if after_sequence.is_some() {
        let deadline = Instant::now() + Duration::from_secs(5);
        let bridge_pid = loop {
            let output = Command::new("pgrep")
                .args(["-P", &child.id().to_string()])
                .output()
                .unwrap();
            if output.status.success() {
                let pid = String::from_utf8(output.stdout).unwrap();
                if let Some(pid) = pid.lines().next() {
                    break pid.to_owned();
                }
            }
            assert!(Instant::now() < deadline, "ACP bridge child did not start");
            std::thread::sleep(Duration::from_millis(10));
        };
        let process = Command::new("ps")
            .args(["eww", "-p", &bridge_pid])
            .output()
            .unwrap();
        assert!(
            !String::from_utf8_lossy(&process.stdout).contains(credential.token),
            "bearer secret appeared in ACP bridge argv/environment"
        );
    }
    let stdout = capture_pipe(child.stdout.take().unwrap());
    let stderr = capture_pipe(child.stderr.take().unwrap());
    let mut stdin = child.stdin.take().unwrap();
    writeln!(stdin, "{prompt}").unwrap();
    stdin.flush().unwrap();

    let deadline = Instant::now() + Duration::from_secs(15);
    loop {
        let last_history = http_rpc_as(
            address,
            credential.token,
            Some(credential.client_id),
            800,
            "history",
            serde_json::json!({"limit":1_000}),
        );
        let history = last_history["result"]["messages"].to_string();
        if history.contains(prompt) && history.contains(expected_reply) {
            break;
        }
        assert!(
            child.try_wait().unwrap().is_none(),
            "tunneled frontend exited before its turn persisted"
        );
        if Instant::now() >= deadline {
            child.kill().ok();
            drop(stdin);
            child.wait().ok();
            let stdout = recv_capture(stdout, "failed tunneled frontend stdout").unwrap();
            let stderr = recv_capture(stderr, "failed tunneled frontend stderr").unwrap();
            panic!(
                "tunneled frontend turn did not persist; history={last_history}; stdout={}; stderr={}",
                String::from_utf8_lossy(&stdout),
                String::from_utf8_lossy(&stderr)
            );
        }
        std::thread::sleep(Duration::from_millis(20));
    }
    drop(stdin);
    let status = wait_child_bounded(&mut child, Duration::from_secs(10), "tunneled frontend");
    Output {
        status,
        stdout: recv_capture(stdout, "tunneled frontend stdout").unwrap(),
        stderr: recv_capture(stderr, "tunneled frontend stderr").unwrap(),
    }
}

#[cfg(unix)]
#[test]
fn remote_instructions_mint_private_fd_credential_and_revoke_on_host_detach() {
    use std::os::unix::fs::{MetadataExt, PermissionsExt};

    let home = temp_home();
    let supercode_home = home.join("supercode-home");
    fs::create_dir_all(&supercode_home).unwrap();
    fs::write(
        supercode_home.join("config.toml"),
        "approval = \"never\"\nsandbox = \"danger-full-access\"\n\
         [core.retry]\nenabled = false\n\
         [capabilities.server]\nenabled = true\n",
    )
    .unwrap();
    let source = home.join("remote-instructions-source.jsonl");
    fs::write(
        &source,
        serde_json::json!({
            "type":"user", "sessionId":"remote-instructions-source", "cwd":home,
            "timestamp":"2020-01-01T00:00:00Z",
            "message":{"role":"user","content":"private remote credential fixture"}
        })
        .to_string()
            + "\n",
    )
    .unwrap();
    let (provider, provider_task) = spawn_echo_provider(0);
    let bootstrap = "remote-instructions-bootstrap";
    let (mut host, address, mut host_stderr) =
        spawn_registry_host(&home, &supercode_home, &source, provider, bootstrap);
    let listed = list_live_sessions(&home, &supercode_home);
    assert_eq!(listed.len(), 1, "{listed:#?}");
    let runtime_id = listed[0]["id"].as_str().unwrap();

    let mut host_attach_command = Command::new(bin());
    host_attach_command
        .env("HOME", &home)
        .env("SUPERCODE_HOME", &supercode_home)
        .env_remove("SUPERCODE_SERVER_TOKEN")
        .args(["--quiet", "attach", runtime_id, "--remote-instructions"])
        .stdin(Stdio::piped())
        .stdout(Stdio::null())
        .stderr(Stdio::piped());
    let mut host_attach = host_attach_command.spawn().unwrap();
    let host_attach_stdin = host_attach.stdin.take().unwrap();
    let host_attach_stderr = host_attach.stderr.take().unwrap();
    set_nonblocking(&host_attach_stderr);
    let mut host_attach_stderr = BufReader::new(host_attach_stderr);
    let deadline = Instant::now() + Duration::from_secs(10);
    let mut diagnostics = String::new();
    let (credential_path, client_id) = loop {
        assert!(
            host_attach.try_wait().unwrap().is_none(),
            "host attach exited before printing remote instructions: {diagnostics}"
        );
        let mut line = String::new();
        match host_attach_stderr.read_line(&mut line) {
            Ok(0) => {}
            Ok(_) => diagnostics.push_str(&line),
            Err(error)
                if matches!(
                    error.kind(),
                    std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted
                ) => {}
            Err(error) => panic!("remote instruction stderr failed: {error}"),
        }
        let path = diagnostics
            .lines()
            .find_map(|line| line.split("'cat ").nth(1)?.strip_suffix("')"))
            .map(PathBuf::from);
        let id = diagnostics
            .lines()
            .find_map(|line| line.split("--client-id ").nth(1))
            .map(str::trim)
            .map(str::to_owned);
        if let (Some(path), Some(id)) = (path, id) {
            break (path, id);
        }
        assert!(
            Instant::now() < deadline,
            "remote instructions timed out: {diagnostics}"
        );
        std::thread::sleep(Duration::from_millis(10));
    };
    let private_dir = fs::symlink_metadata(credential_path.parent().unwrap()).unwrap();
    let private_file = fs::symlink_metadata(&credential_path).unwrap();
    assert!(private_dir.is_dir() && !private_dir.file_type().is_symlink());
    assert_eq!(private_dir.permissions().mode() & 0o777, 0o700);
    assert_eq!(private_dir.uid(), unsafe { libc::geteuid() });
    assert!(private_file.is_file() && !private_file.file_type().is_symlink());
    assert_eq!(private_file.permissions().mode() & 0o777, 0o600);
    assert_eq!(private_file.uid(), unsafe { libc::geteuid() });
    let scoped = fs::read_to_string(&credential_path).unwrap();
    assert_eq!(scoped.len(), 64);
    assert!(!diagnostics.contains(&scoped));
    assert!(!host_attach_command
        .get_args()
        .any(|argument| argument.to_string_lossy().contains(&scoped)));

    let credential_file = fs::File::open(&credential_path).unwrap();
    let credential_fd = credential_file.as_raw_fd();
    let inherited_fd = 9;
    let mut client_command = Command::new(bin());
    client_command
        .env("HOME", &home)
        .env("SUPERCODE_HOME", &supercode_home)
        .env_remove("SUPERCODE_SERVER_TOKEN")
        .args([
            "--quiet",
            "attach",
            "--connect",
            &format!("http://{address}"),
            "--credential-fd",
            &inherited_fd.to_string(),
            "--client-id",
            &client_id,
        ])
        .stdin(Stdio::piped())
        .stdout(Stdio::null())
        .stderr(Stdio::piped());
    // SAFETY: `credential_fd` remains open through spawn. `dup2` creates the
    // inherited descriptor requested by the child and clears CLOEXEC on it.
    unsafe {
        client_command.pre_exec(move || {
            if libc::dup2(credential_fd, inherited_fd) == -1 {
                return Err(std::io::Error::last_os_error());
            }
            Ok(())
        });
    }
    let visible_args = client_command
        .get_args()
        .map(|argument| argument.to_string_lossy())
        .collect::<Vec<_>>();
    assert!(visible_args
        .iter()
        .all(|argument| !argument.contains(&scoped)));
    let mut client = client_command.spawn().unwrap();
    let client_stdin = client.stdin.take().unwrap();
    let client_stderr = capture_pipe(client.stderr.take().unwrap());
    let client_pid = client.id();
    std::thread::sleep(Duration::from_millis(100));
    assert!(client.try_wait().unwrap().is_none());
    let process = Command::new("ps")
        .args(["eww", "-p", &client_pid.to_string()])
        .output()
        .unwrap();
    assert!(
        !String::from_utf8_lossy(&process.stdout).contains(&scoped),
        "scoped credential appeared in client argv/environment"
    );
    drop(client_stdin);
    let client_status = wait_child_bounded(&mut client, Duration::from_secs(10), "FD frontend");
    let client_diagnostics = recv_capture(client_stderr, "FD frontend stderr").unwrap();
    assert!(
        client_status.success(),
        "{}",
        String::from_utf8_lossy(&client_diagnostics)
    );
    assert!(!String::from_utf8_lossy(&client_diagnostics).contains(&scoped));

    let describe = http_rpc_as(
        address,
        &scoped,
        Some(&client_id),
        1,
        "frontend.describe",
        serde_json::json!({}),
    );
    assert_eq!(describe["result"]["session_id"], runtime_id);
    let close = http_rpc_as(
        address,
        &scoped,
        Some(&client_id),
        2,
        "frontend.close",
        serde_json::json!({}),
    );
    assert_eq!(close["error"]["name"], "unauthorized");
    assert_eq!(close["error"]["permission"], "terminate");
    assert!(host.try_wait().unwrap().is_none());

    drop(host_attach_stdin);
    assert!(wait_child_bounded(
        &mut host_attach,
        Duration::from_secs(10),
        "remote instruction host attach"
    )
    .success());
    read_to_string_bounded(
        &mut host_attach_stderr,
        &mut diagnostics,
        Duration::from_secs(2),
        "remote instruction host attach diagnostics",
    );
    assert!(!diagnostics.contains(&scoped));
    assert!(!credential_path.exists());
    let revoked = http_rpc_as(
        address,
        &scoped,
        Some(&client_id),
        3,
        "frontend.describe",
        serde_json::json!({}),
    );
    assert_eq!(revoked["error"]["name"], "unauthenticated");
    assert!(host.try_wait().unwrap().is_none());

    shutdown(address, bootstrap);
    assert!(wait_child_bounded(&mut host, Duration::from_secs(10), "runtime host").success());
    let mut runtime_diagnostics = String::new();
    read_to_string_bounded(
        &mut host_stderr,
        &mut runtime_diagnostics,
        Duration::from_secs(2),
        "remote instruction runtime diagnostics",
    );
    assert!(!runtime_diagnostics.contains(&scoped));
    assert!(!runtime_diagnostics.contains(bootstrap));
    join_thread_bounded(provider_task, Duration::from_secs(2), "unused provider");
    fs::remove_dir_all(home).ok();
}

#[cfg(unix)]
#[test]
fn three_process_registry_coordinates_http_acp_and_legacy_clients() {
    let home = temp_home();
    let supercode_home = home.join("supercode-home");
    fs::create_dir_all(&supercode_home).unwrap();
    fs::write(
        supercode_home.join("config.toml"),
        "approval = \"never\"\nsandbox = \"danger-full-access\"\n\
         [core.retry]\nenabled = false\n\
         [capabilities.server]\nenabled = true\n",
    )
    .unwrap();
    let mut sources = Vec::new();
    for index in 0..3 {
        let path = home.join(format!("registry-source-{index}.jsonl"));
        let bytes = serde_json::json!({
            "type": "user",
            "sessionId": format!("registry-source-{index}"),
            "cwd": home,
            "timestamp": "2020-01-01T00:00:00Z",
            "message": {"role": "user", "content": format!("source fact {index}")}
        })
        .to_string()
            + "\n";
        fs::write(&path, &bytes).unwrap();
        sources.push((path, bytes.into_bytes()));
    }
    let (provider, provider_task) = spawn_echo_provider(6);
    let tokens = ["registry-token-0", "registry-token-1", "registry-token-2"];
    let mut hosts = Vec::new();
    for index in 0..3 {
        hosts.push(spawn_registry_host(
            &home,
            &supercode_home,
            &sources[index].0,
            provider,
            tokens[index],
        ));
    }

    let listed = list_live_sessions(&home, &supercode_home);
    assert_eq!(listed.len(), 3, "{listed:#?}");

    let mut alpha_events = open_frontend_events(hosts[0].1, tokens[0], "alpha");
    let mut beta_events = open_frontend_events(hosts[0].1, tokens[0], "beta");
    let alpha = http_rpc_as(
        hosts[0].1,
        tokens[0],
        Some("alpha"),
        1,
        "submit",
        serde_json::json!({"prompt":"REGISTRY_HTTP_ALPHA"}),
    );
    assert!(alpha["result"]["reply"]
        .as_str()
        .unwrap()
        .starts_with("REGISTRY REPLY"));
    let denied = http_rpc_as(
        hosts[0].1,
        tokens[0],
        Some("beta"),
        2,
        "submit",
        serde_json::json!({"prompt":"REGISTRY_IMPLICIT_TAKEOVER"}),
    );
    assert_eq!(denied["error"]["name"], "controller_required");
    assert_eq!(denied["error"]["holder"], "alpha");
    let takeover = http_rpc_as(
        hosts[0].1,
        tokens[0],
        Some("beta"),
        3,
        "frontend.take_control",
        serde_json::json!({}),
    );
    assert_eq!(takeover["result"]["controller"]["client_id"], "beta");
    assert!(http_rpc_as(
        hosts[0].1,
        tokens[0],
        Some("beta"),
        4,
        "submit",
        serde_json::json!({"prompt":"REGISTRY_HTTP_BETA"}),
    )["result"]["reply"]
        .as_str()
        .unwrap()
        .starts_with("REGISTRY REPLY"));
    let stale = http_rpc_as(
        hosts[0].1,
        tokens[0],
        Some("lost-controller"),
        5,
        "frontend.take_control",
        serde_json::json!({}),
    );
    assert_eq!(
        stale["result"]["controller"]["client_id"],
        "lost-controller"
    );
    std::thread::sleep(Duration::from_millis(140));
    let expired = http_rpc_as(
        hosts[0].1,
        tokens[0],
        Some("lost-controller"),
        6,
        "submit",
        serde_json::json!({"prompt":"REGISTRY_EXPIRED_CLIENT_MUST_NOT_RUN"}),
    );
    assert_eq!(expired["error"]["name"], "lease_expired", "{expired:#}");
    let recovered = http_rpc_as(
        hosts[0].1,
        tokens[0],
        Some("alpha"),
        7,
        "submit",
        serde_json::json!({"prompt":"REGISTRY_AFTER_CLIENT_LOSS"}),
    );
    assert!(recovered["result"]["reply"]
        .as_str()
        .unwrap()
        .starts_with("REGISTRY REPLY"));
    let recovered_lease = http_rpc_as(
        hosts[0].1,
        tokens[0],
        Some("alpha"),
        9,
        "frontend.lease",
        serde_json::json!({}),
    );
    assert_eq!(
        recovered_lease["result"]["controller"]["client_id"],
        "alpha"
    );
    assert_eq!(
        recovered_lease["result"]["observers"]
            .as_array()
            .unwrap()
            .len(),
        3
    );
    let alpha_observed = read_frontend_events_until(&mut alpha_events, 3);
    let beta_observed = read_frontend_events_until(&mut beta_events, 3);
    assert_eq!(alpha_observed, beta_observed);
    assert!(alpha_observed.windows(2).all(|events| {
        events[0]["sequence"].as_u64().unwrap() < events[1]["sequence"].as_u64().unwrap()
    }));

    let mut bridge = Command::new(bin())
        .env("HOME", &home)
        .env("SUPERCODE_HOME", &supercode_home)
        .args([
            "--quiet",
            "acp",
            "--connect",
            &format!("http://{}", hosts[1].1),
            "--token",
            tokens[1],
        ])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let mut bridge_stdin = bridge.stdin.take().unwrap();
    let bridge_stdout = bridge.stdout.take().unwrap();
    set_nonblocking(&bridge_stdout);
    let mut bridge_stdout = BufReader::new(bridge_stdout);
    send(
        &mut bridge_stdin,
        serde_json::json!({
            "jsonrpc":"2.0","id":1,"method":"initialize",
            "params":{"protocolVersion":1,"clientCapabilities":{}}
        }),
    );
    assert_eq!(receive(&mut bridge_stdout)["result"]["protocolVersion"], 1);
    send(
        &mut bridge_stdin,
        serde_json::json!({
            "jsonrpc":"2.0","id":2,"method":"session/new",
            "params":{"cwd":home,"mcpServers":[]}
        }),
    );
    let acp_session = receive(&mut bridge_stdout)["result"]["sessionId"]
        .as_str()
        .unwrap()
        .to_string();
    send(
        &mut bridge_stdin,
        serde_json::json!({
            "jsonrpc":"2.0","id":3,"method":"session/prompt",
            "params":{"sessionId":acp_session,"prompt":[{"type":"text","text":"REGISTRY_ACP"}]}
        }),
    );
    loop {
        let value = receive(&mut bridge_stdout);
        if value["id"] == 3 {
            assert_eq!(value["result"]["stopReason"], "end_turn");
            break;
        }
    }
    send(
        &mut bridge_stdin,
        serde_json::json!({
            "jsonrpc":"2.0","id":4,"method":"supercode/frontend/detach",
            "params":{"sessionId":acp_session}
        }),
    );
    loop {
        let value = receive(&mut bridge_stdout);
        if value["id"] == 4 {
            assert!(value.get("result").is_some(), "{value}");
            break;
        }
        assert!(
            value.get("method").is_some(),
            "unexpected ACP frame before detach response: {value}"
        );
    }
    drop(bridge_stdin);
    assert!(wait_child_bounded(&mut bridge, Duration::from_secs(10), "registry ACP").success());

    let legacy = http_rpc(
        hosts[2].1,
        tokens[2],
        5,
        "submit",
        serde_json::json!({"prompt":"REGISTRY_LEGACY"}),
    );
    assert!(legacy["result"]["reply"]
        .as_str()
        .unwrap()
        .starts_with("REGISTRY REPLY"));
    let history_before_restart = http_rpc(
        hosts[2].1,
        tokens[2],
        51,
        "history",
        serde_json::json!({"limit":1000}),
    )["result"]["messages"]
        .clone();

    let listed = list_live_sessions(&home, &supercode_home);
    let first = listed
        .iter()
        .find(|entry| entry["source_session_id"] == "registry-source-0")
        .unwrap();
    assert_eq!(first["observers"].as_array().unwrap().len(), 3);
    assert!(listed.iter().all(|entry| entry["state"] == "idle"));
    for entry in &listed {
        let path = entry["persistence_location"].as_str().unwrap();
        assert!(std::path::Path::new(path).exists(), "missing {path}");
    }

    let host_two_entry = listed
        .iter()
        .find(|entry| entry["source_session_id"] == "registry-source-2")
        .unwrap();
    let first_persistence = PathBuf::from(host_two_entry["persistence_location"].as_str().unwrap());
    shutdown(hosts[2].1, tokens[2]);
    assert!(wait_child_bounded(
        &mut hosts[2].0,
        Duration::from_secs(10),
        "registry host before restart"
    )
    .success());
    let first_persistence_before_restart = fs::read(&first_persistence).unwrap();
    let first_history = String::from_utf8(first_persistence_before_restart.clone()).unwrap();
    assert_eq!(first_history.matches("REGISTRY_LEGACY").count(), 1);
    let first_persisted_messages = first_history
        .lines()
        .filter(|line| !line.trim().is_empty())
        .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
        .collect::<Vec<_>>();
    assert_eq!(
        serde_json::Value::Array(first_persisted_messages),
        history_before_restart
    );
    let restart_deadline = Instant::now() + Duration::from_secs(5);
    loop {
        let current = list_live_sessions(&home, &supercode_home);
        if current.len() == 2 {
            break;
        }
        assert!(
            Instant::now() < restart_deadline,
            "closed runtime receipt did not reconcile: {current:#?}"
        );
        std::thread::sleep(Duration::from_millis(25));
    }
    hosts[2] = spawn_registry_host(
        &home,
        &supercode_home,
        &first_persistence,
        provider,
        tokens[2],
    );
    let restarted = http_rpc(
        hosts[2].1,
        tokens[2],
        8,
        "submit",
        serde_json::json!({"prompt":"REGISTRY_AFTER_RESTART"}),
    );
    assert!(restarted["result"]["reply"]
        .as_str()
        .unwrap()
        .starts_with("REGISTRY REPLY"));
    let history_after_restart = http_rpc(
        hosts[2].1,
        tokens[2],
        52,
        "history",
        serde_json::json!({"limit":1000}),
    )["result"]["messages"]
        .clone();
    let before = history_before_restart.as_array().unwrap();
    let after = history_after_restart.as_array().unwrap();
    assert_eq!(&after[..before.len()], before);
    let after_restart = list_live_sessions(&home, &supercode_home);
    assert_eq!(after_restart.len(), 3, "{after_restart:#?}");
    assert!(after_restart.iter().all(|entry| entry["state"] == "idle"));
    let restarted_entry = after_restart
        .iter()
        .find(|entry| {
            entry["source_session_id"] != "registry-source-0"
                && entry["source_session_id"] != "registry-source-1"
        })
        .expect("restarted runtime");
    assert_eq!(restarted_entry["source_harness"], "claude-code");
    let restarted_persistence =
        PathBuf::from(restarted_entry["persistence_location"].as_str().unwrap());

    let _ = http_rpc_as(
        hosts[0].1,
        tokens[0],
        Some("beta"),
        90,
        "frontend.close",
        serde_json::json!({}),
    );
    shutdown(hosts[1].1, tokens[1]);
    shutdown(hosts[2].1, tokens[2]);
    for (host, _, _) in &mut hosts {
        assert!(wait_child_bounded(host, Duration::from_secs(10), "registry host").success());
    }
    join_thread_bounded(provider_task, Duration::from_secs(10), "registry provider");
    assert_eq!(
        fs::read(&first_persistence).unwrap(),
        first_persistence_before_restart
    );
    let restarted_history = fs::read_to_string(restarted_persistence).unwrap();
    assert_eq!(restarted_history.matches("REGISTRY_LEGACY").count(), 1);
    assert_eq!(
        restarted_history.matches("REGISTRY_AFTER_RESTART").count(),
        1
    );
    assert_eq!(
        restarted_history
            .lines()
            .filter(|line| line.contains("REGISTRY REPLY"))
            .count(),
        2,
        "restart must preserve one pre-restart reply and append one new reply"
    );
    let restarted_persisted_messages = restarted_history
        .lines()
        .filter(|line| !line.trim().is_empty())
        .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
        .collect::<Vec<_>>();
    assert_eq!(
        serde_json::Value::Array(restarted_persisted_messages),
        history_after_restart
    );
    for (path, before) in &sources {
        assert_eq!(&fs::read(path).unwrap(), before);
    }
    assert!(list_live_sessions(&home, &supercode_home).is_empty());
    fs::remove_dir_all(home).ok();
}

/// Hosted proof for SUP-65's separately packaged Pi-derived display client.
/// The release/compatibility job supplies Node and the package dependencies;
/// ordinary Rust workspace tests stay hermetic.
#[cfg(unix)]
#[test]
#[ignore = "requires npm ci --prefix sdk/frontend-pi, Node 22.19+, and tmux"]
fn packaged_pi_frontend_drives_one_canonical_runtime_and_detaches_cleanly() {
    let repository = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..");
    let pi_entrypoint = repository.join("sdk/frontend-pi/bin/supercode-pi.mjs");
    assert!(
        pi_entrypoint.exists(),
        "run npm ci --prefix sdk/frontend-pi"
    );
    assert!(repository.join("sdk/frontend-pi/node_modules").exists());

    let home = temp_home();
    let pi_bin = home.join("pi-bin");
    fs::create_dir(&pi_bin).unwrap();
    std::os::unix::fs::symlink(&pi_entrypoint, pi_bin.join("supercode-pi")).unwrap();
    let supercode_home = home.join("supercode-home");
    fs::create_dir_all(&supercode_home).unwrap();
    fs::write(
        supercode_home.join("config.toml"),
        "approval = \"never\"\nsandbox = \"danger-full-access\"\n\
         [core.retry]\nenabled = false\n\
         [capabilities.server]\nenabled = true\n\
         [capabilities.tui]\nenabled = true\n",
    )
    .unwrap();
    let source = home.join("pi-source.jsonl");
    let source_bytes = serde_json::json!({
        "type": "user",
        "sessionId": "pi-source",
        "cwd": home,
        "timestamp": "2020-01-01T00:00:00Z",
        "message": {"role": "user", "content": "PI_IMPORTED_FACT"}
    })
    .to_string()
        + "\n";
    fs::write(&source, &source_bytes).unwrap();
    let (provider, provider_task) = spawn_echo_provider(1);
    let bootstrap = "pi-proof-bootstrap";
    let (mut host, address, mut host_stderr) =
        spawn_registry_host(&home, &supercode_home, &source, provider, bootstrap);
    let live = list_live_sessions(&home, &supercode_home);
    assert_eq!(live.len(), 1, "{live:#?}");
    let runtime_id = live[0]["runtime_id"].as_str().unwrap().to_string();
    let persistence = PathBuf::from(live[0]["persistence_location"].as_str().unwrap());

    let tmux = TmuxServer(format!("supercode-sup65-pi-{}", std::process::id()));
    let created = tmux.command(&[
        "new-session",
        "-d",
        "-s",
        "frontend",
        "-x",
        "120",
        "-y",
        "40",
    ]);
    assert!(created.status.success(), "{:?}", created.stderr);
    tmux.send_line("echo SC_PI_INITIAL_SHELL_READY");
    tmux.expect_count("SC_PI_INITIAL_SHELL_READY", 2);
    std::thread::sleep(Duration::from_millis(100));
    let original_termios = tmux.termios();
    let leak_log = home.join("pi-inherited-leak.log");
    let attach = format!(
        "PATH={}:$PATH PI_DEBUG_REDRAW=1 PI_TUI_DEBUG=1 PI_TUI_WRITE_LOG={} HOME={} SUPERCODE_HOME={} TERM=xterm-256color {} attach {} --ui pi --take-control; echo SC_PI_DONE:$?",
        shell_word(pi_bin.to_str().unwrap()),
        shell_word(leak_log.to_str().unwrap()),
        shell_word(home.to_str().unwrap()),
        shell_word(supercode_home.to_str().unwrap()),
        shell_word(bin()),
        shell_word(&runtime_id),
    );
    tmux.send_line(&attach);
    tmux.expect("frontend availability: id=pi installed=true compatible=true");
    tmux.expect("PI_IMPORTED_FACT");
    tmux.send_tui_line("REGISTRY_PI_CANONICAL_PROMPT");
    tmux.expect("REGISTRY REPLY 0");
    let detach = tmux.command(&["send-keys", "-t", "frontend", "C-d"]);
    assert!(detach.status.success(), "{:?}", detach.stderr);
    tmux.expect("SC_PI_DONE:0");
    tmux.send_line("echo SC_PI_SHELL_READY");
    tmux.expect_count("SC_PI_SHELL_READY", 2);
    std::thread::sleep(Duration::from_millis(100));
    assert_eq!(tmux.termios(), original_termios);
    assert!(!leak_log.exists());
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 1);

    let history = http_rpc(
        address,
        bootstrap,
        65,
        "history",
        serde_json::json!({"limit":1000}),
    )["result"]["messages"]
        .clone();
    let wire = history.to_string();
    assert_eq!(wire.matches("PI_IMPORTED_FACT").count(), 1, "{wire}");
    assert_eq!(
        wire.matches("REGISTRY_PI_CANONICAL_PROMPT").count(),
        1,
        "{wire}"
    );
    assert_eq!(wire.matches("REGISTRY REPLY 0").count(), 1, "{wire}");
    assert_eq!(fs::read_to_string(&source).unwrap(), source_bytes);
    assert_eq!(
        fs::read_to_string(&persistence)
            .unwrap()
            .matches("REGISTRY_PI_CANONICAL_PROMPT")
            .count(),
        1
    );

    shutdown(address, bootstrap);
    assert!(wait_child_bounded(&mut host, Duration::from_secs(10), "Pi runtime").success());
    let mut diagnostics = String::new();
    read_to_string_bounded(
        &mut host_stderr,
        &mut diagnostics,
        Duration::from_secs(2),
        "Pi runtime diagnostics",
    );
    assert!(!diagnostics.contains(bootstrap));
    join_thread_bounded(provider_task, Duration::from_secs(10), "Pi provider");
    fs::remove_dir_all(home).ok();
}

/// Real-terminal proof for the `TERM=dumb` fallback. The independently packed
/// client must retain Pi's semantic input contract and restore the caller's
/// exact terminal state after both detach and runtime close.
#[cfg(unix)]
#[test]
#[ignore = "requires SUPERCODE_PACKED_PI_BIN and tmux"]
fn dumb_pi_frontend_preserves_real_pty_input_and_terminal_lifecycle() {
    let repository = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..");
    let pi_executable = PathBuf::from(
        std::env::var("SUPERCODE_PACKED_PI_BIN")
            .expect("set SUPERCODE_PACKED_PI_BIN to an isolated npm-packed install"),
    );
    let pi_version = Command::new(&pi_executable)
        .arg("--version")
        .output()
        .unwrap();
    assert!(pi_version.status.success(), "{:?}", pi_version.stderr);
    assert_eq!(
        String::from_utf8_lossy(&pi_version.stdout).trim(),
        "supercode-pi 0.1.0"
    );
    let pi_executable = fs::canonicalize(pi_executable).unwrap();
    assert!(
        !pi_executable.starts_with(fs::canonicalize(repository).unwrap()),
        "dumb-terminal proof must execute an isolated packed install: {}",
        pi_executable.display()
    );

    let home = temp_home();
    let supercode_home = home.join("supercode-home");
    fs::create_dir_all(&supercode_home).unwrap();
    fs::write(
        supercode_home.join("config.toml"),
        "approval = \"never\"\nsandbox = \"danger-full-access\"\n\
         [core.retry]\nenabled = false\n\
         [capabilities.server]\nenabled = true\n\
         [capabilities.tui]\nenabled = true\n",
    )
    .unwrap();
    let source = home.join("dumb-source.jsonl");
    let source_bytes = serde_json::json!({
        "parentUuid": null,
        "type": "user",
        "uuid": "30000000-0000-4000-8000-000000000000",
        "sessionId": "dumb-source",
        "cwd": home,
        "timestamp": "2020-01-01T00:00:00Z",
        "message": {"role": "user", "content": "DUMB_IMPORTED_FACT"}
    })
    .to_string()
        + "\n";
    fs::write(&source, &source_bytes).unwrap();

    let (provider, provider_task, interrupt_started) = spawn_dumb_terminal_provider();
    let bootstrap = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd";
    let (mut host, address, _stderr) =
        spawn_registry_host_with_ttl(&home, &supercode_home, &source, provider, bootstrap, 30_000);

    let tmux = TmuxServer(format!("supercode-sup65-dumb-{}", std::process::id()));
    let created = tmux.command(&[
        "new-session",
        "-d",
        "-s",
        "frontend",
        "-x",
        "80",
        "-y",
        "24",
    ]);
    assert!(created.status.success(), "{:?}", created.stderr);
    tmux.send_line("echo SC_DUMB_SHELL_0");
    tmux.expect_count("SC_DUMB_SHELL_0", 2);
    std::thread::sleep(Duration::from_millis(100));
    let original_termios = tmux.termios();

    let first_credential = home.join("dumb-first.credential");
    write_private_credential(&first_credential, bootstrap);
    let first_attach = format!(
        "HOME={} TERM=dumb SUPERCODE_PI_FRONTEND_URL={} SUPERCODE_PI_FRONTEND_CLIENT_ID=dumb-owner-1 SUPERCODE_PI_FRONTEND_CREDENTIAL_FILE={} SUPERCODE_PI_FRONTEND_PERMISSIONS=observe,interact,approve,terminate SUPERCODE_PI_FRONTEND_TAKE_CONTROL=1 {}; echo SC_DUMB_FIRST_DONE:$?",
        shell_word(home.to_str().unwrap()),
        shell_word(&format!("http://{address}")),
        shell_word(first_credential.to_str().unwrap()),
        shell_word(pi_executable.to_str().unwrap()),
    );
    tmux.send_line(&first_attach);
    tmux.expect("DUMB_IMPORTED_FACT");
    tmux.expect("connected · idle");
    tmux.send_tui_line("DUMB_SUBMIT_PROMPT");
    tmux.expect("DUMB SUBMIT DONE");
    tmux.send_tui_line("DUMB_INTERRUPT_PROMPT");
    interrupt_started
        .recv_timeout(Duration::from_secs(10))
        .expect("dumb-terminal interrupt prompt reached provider");
    let interrupt = tmux.command(&["send-keys", "-t", "frontend", "C-c"]);
    assert!(interrupt.status.success(), "{:?}", interrupt.stderr);
    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        let state = http_rpc(
            address,
            bootstrap,
            66,
            "frontend.v2.describe",
            serde_json::json!({}),
        )["result"]["turn_state"]
            .as_str()
            .unwrap()
            .to_string();
        if state == "idle" {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "Ctrl-C did not interrupt the turn"
        );
        std::thread::sleep(Duration::from_millis(20));
    }
    let detach = tmux.command(&["send-keys", "-t", "frontend", "C-d"]);
    assert!(detach.status.success(), "{:?}", detach.stderr);
    tmux.expect("SC_DUMB_FIRST_DONE:0");
    tmux.send_line("echo SC_DUMB_SHELL_1");
    tmux.expect_count("SC_DUMB_SHELL_1", 2);
    std::thread::sleep(Duration::from_millis(100));
    assert_eq!(tmux.termios(), original_termios);
    assert!(!first_credential.exists());

    let history = http_rpc(
        address,
        bootstrap,
        67,
        "history",
        serde_json::json!({"limit": 1000}),
    )["result"]["messages"]
        .clone();
    let history_wire = history.to_string();
    for marker in [
        "DUMB_IMPORTED_FACT",
        "DUMB_SUBMIT_PROMPT",
        "DUMB SUBMIT DONE",
        "DUMB_INTERRUPT_PROMPT",
    ] {
        assert_eq!(history_wire.matches(marker).count(), 1, "{history_wire}");
    }
    assert_eq!(fs::read_to_string(&source).unwrap(), source_bytes);

    let close_credential = home.join("dumb-close.credential");
    write_private_credential(&close_credential, bootstrap);
    let close_attach = format!(
        "HOME={} TERM=dumb SUPERCODE_PI_FRONTEND_URL={} SUPERCODE_PI_FRONTEND_CLIENT_ID=dumb-owner-2 SUPERCODE_PI_FRONTEND_CREDENTIAL_FILE={} SUPERCODE_PI_FRONTEND_PERMISSIONS=observe,interact,approve,terminate SUPERCODE_PI_FRONTEND_TAKE_CONTROL=1 {}; echo SC_DUMB_CLOSE_DONE:$?",
        shell_word(home.to_str().unwrap()),
        shell_word(&format!("http://{address}")),
        shell_word(close_credential.to_str().unwrap()),
        shell_word(pi_executable.to_str().unwrap()),
    );
    tmux.send_line(&close_attach);
    tmux.expect("connected · idle");
    let close = tmux.command(&["send-keys", "-t", "frontend", "C-x"]);
    assert!(close.status.success(), "{:?}", close.stderr);
    tmux.expect("SC_DUMB_CLOSE_DONE:0");
    tmux.send_line("echo SC_DUMB_SHELL_2");
    tmux.expect_count("SC_DUMB_SHELL_2", 2);
    std::thread::sleep(Duration::from_millis(100));
    assert_eq!(tmux.termios(), original_termios);
    assert!(!close_credential.exists());

    assert!(
        wait_child_bounded(&mut host, Duration::from_secs(10), "closed dumb runtime").success()
    );
    join_thread_bounded(
        provider_task,
        Duration::from_secs(10),
        "dumb-terminal provider",
    );
    tmux.command(&["kill-server"]);
    fs::remove_dir_all(home).ok();
}

/// End-to-end release proof for SUP-65's frontend interchange contract. Two
/// unmodified, separately packaged display clients take turns driving the
/// same imported Claude-origin SDK runtime in one real tmux pane. Neither
/// client owns canonical history, persistence, export, or runtime lifetime.
#[cfg(unix)]
#[test]
#[ignore = "requires unmodified codex-cli 0.144.4, SUPERCODE_PACKED_PI_BIN, and tmux"]
fn stock_codex_pi_stock_codex_swap_preserves_one_exportable_runtime() {
    let version = Command::new("codex").arg("--version").output().unwrap();
    assert!(version.status.success());
    assert_eq!(
        String::from_utf8_lossy(&version.stdout).trim(),
        "codex-cli 0.144.4"
    );
    let codex_path =
        output_bounded(Command::new("which").arg("codex"), Duration::from_secs(3)).unwrap();
    assert!(codex_path.status.success(), "{:?}", codex_path.stderr);
    let codex_path = PathBuf::from(String::from_utf8(codex_path.stdout).unwrap().trim());
    let codex_bin = codex_path.parent().unwrap();
    let tmux_version = Command::new("tmux").arg("-V").output().unwrap();
    assert!(tmux_version.status.success());

    let repository = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..");
    let pi_executable = PathBuf::from(
        std::env::var("SUPERCODE_PACKED_PI_BIN")
            .expect("set SUPERCODE_PACKED_PI_BIN to an isolated npm-packed install"),
    );
    let pi_version = Command::new(&pi_executable)
        .arg("--version")
        .output()
        .unwrap();
    assert!(pi_version.status.success(), "{:?}", pi_version.stderr);
    assert_eq!(
        String::from_utf8_lossy(&pi_version.stdout).trim(),
        "supercode-pi 0.1.0"
    );
    let pi_bin = pi_executable.parent().unwrap().to_path_buf();
    let pi_executable = fs::canonicalize(pi_executable).unwrap();
    assert!(
        !pi_executable.starts_with(fs::canonicalize(&repository).unwrap()),
        "interchange proof must execute an isolated packed install, not repository source: {}",
        pi_executable.display()
    );

    let home = temp_home();
    let supercode_home = home.join("supercode-home");
    fs::create_dir_all(&supercode_home).unwrap();
    fs::write(
        supercode_home.join("config.toml"),
        "approval = \"never\"\nsandbox = \"danger-full-access\"\n\
         [core.retry]\nenabled = false\n\
         [capabilities.server]\nenabled = true\n\
         [capabilities.tui]\nenabled = true\n",
    )
    .unwrap();
    let source = home.join("interchange-source.jsonl");
    let source_bytes = serde_json::json!({
        "parentUuid": null,
        "type": "user",
        "uuid": "10000000-0000-4000-8000-000000000000",
        "sessionId": "interchange-source",
        "cwd": home,
        "timestamp": "2020-01-01T00:00:00Z",
        "message": {"role": "user", "content": "INTERCHANGE_IMPORTED_FACT"}
    })
    .to_string()
        + "\n";
    fs::write(&source, &source_bytes).unwrap();

    let (provider, provider_task) = spawn_echo_provider(2);
    let token = "interchange-proof-token";
    let (mut host, address, _stderr) =
        spawn_registry_host_with_ttl(&home, &supercode_home, &source, provider, token, 30_000);
    let live = list_live_sessions(&home, &supercode_home);
    assert_eq!(live.len(), 1, "{live:#?}");
    let runtime_id = live[0]["runtime_id"].as_str().unwrap().to_string();
    let persistence = PathBuf::from(live[0]["persistence_location"].as_str().unwrap());

    let tmux = TmuxServer(format!(
        "supercode-sup65-interchange-{}",
        std::process::id()
    ));
    let created = tmux.command(&[
        "new-session",
        "-d",
        "-s",
        "frontend",
        "-x",
        "120",
        "-y",
        "40",
    ]);
    assert!(created.status.success(), "{:?}", created.stderr);
    tmux.send_line("echo SC_INTERCHANGE_SHELL_0");
    tmux.expect_count("SC_INTERCHANGE_SHELL_0", 2);
    std::thread::sleep(Duration::from_millis(100));
    let original_termios = tmux.termios();

    // Stock Codex owns the first turn, then its isolated process group is
    // deliberately killed. The host must detach the dead lease and restore
    // this exact pane without ending or rewriting the SDK runtime.
    let codex_attach = format!(
        "PATH={}:$PATH HOME={} SUPERCODE_HOME={} TERM=xterm-256color {} attach {} --ui codex --take-control; echo SC_CODEX_FIRST_DONE:$?",
        shell_word(codex_bin.to_str().unwrap()),
        shell_word(home.to_str().unwrap()),
        shell_word(supercode_home.to_str().unwrap()),
        shell_word(bin()),
        shell_word(&runtime_id),
    );
    tmux.send_line(&codex_attach);
    tmux.expect("frontend availability: id=codex");
    tmux.expect("INTERCHANGE_IMPORTED_FACT");
    tmux.send_line("REGISTRY_CODEX_BEFORE_PI");
    tmux.expect("REGISTRY REPLY 0");
    let attach_pid = only_child_pid(tmux.pane_pid());
    let stock_pid = only_child_pid(attach_pid as u32);
    unsafe { libc::kill(-stock_pid, libc::SIGKILL) };
    tmux.expect("SC_CODEX_FIRST_DONE:");
    tmux.send_line("echo SC_INTERCHANGE_SHELL_1");
    tmux.expect_count("SC_INTERCHANGE_SHELL_1", 2);
    std::thread::sleep(Duration::from_millis(100));
    assert_eq!(tmux.termios(), original_termios);
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 1);
    tmux.clear_screen_and_history();

    // The packaged Pi display takes over the same runtime in the same pane.
    // Ctrl-D is Pi's real detach path; it must return to the original shell.
    let pi_attach = format!(
        "PATH={}:$PATH HOME={} SUPERCODE_HOME={} TERM=xterm-256color {} attach {} --ui pi --take-control; echo SC_PI_MIDDLE_DONE:$?",
        shell_word(pi_bin.to_str().unwrap()),
        shell_word(home.to_str().unwrap()),
        shell_word(supercode_home.to_str().unwrap()),
        shell_word(bin()),
        shell_word(&runtime_id),
    );
    tmux.send_line(&pi_attach);
    tmux.expect("frontend availability: id=pi installed=true compatible=true");
    tmux.expect("connected · idle");
    tmux.expect("REGISTRY REPLY 0");
    tmux.send_tui_line("REGISTRY_PI_MIDDLE");
    tmux.expect("REGISTRY REPLY 1");
    let pi_detach = tmux.command(&["send-keys", "-t", "frontend", "C-d"]);
    assert!(pi_detach.status.success(), "{:?}", pi_detach.stderr);
    tmux.expect("SC_PI_MIDDLE_DONE:0");
    tmux.send_line("echo SC_INTERCHANGE_SHELL_2");
    tmux.expect_count("SC_INTERCHANGE_SHELL_2", 2);
    std::thread::sleep(Duration::from_millis(100));
    assert_eq!(tmux.termios(), original_termios);
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 1);
    tmux.clear_screen_and_history();

    let history_after_turns = http_rpc(
        address,
        token,
        65,
        "history",
        serde_json::json!({"limit":1000}),
    )["result"]["messages"]
        .clone();
    let history_wire = history_after_turns.to_string();
    for marker in [
        "INTERCHANGE_IMPORTED_FACT",
        "REGISTRY_CODEX_BEFORE_PI",
        "REGISTRY REPLY 0",
        "REGISTRY_PI_MIDDLE",
        "REGISTRY REPLY 1",
    ] {
        assert_eq!(history_wire.matches(marker).count(), 1, "{history_wire}");
    }
    let persisted_after_turns = fs::read(&persistence).unwrap();
    let persisted_wire = String::from_utf8(persisted_after_turns.clone()).unwrap();
    for marker in [
        "INTERCHANGE_IMPORTED_FACT",
        "REGISTRY_CODEX_BEFORE_PI",
        "REGISTRY REPLY 0",
        "REGISTRY_PI_MIDDLE",
        "REGISTRY REPLY 1",
    ] {
        assert_eq!(
            persisted_wire.matches(marker).count(),
            1,
            "{persisted_wire}"
        );
    }
    assert_eq!(fs::read_to_string(&source).unwrap(), source_bytes);

    // Export through the canonical saved-session identity, not the flat
    // persistence path. This is the full-fidelity path that rejoins the
    // original Claude body with the two native turns for stock resumption.
    let export_before = home.join("before-final-codex.jsonl");
    let mut convert_before = Command::new(bin());
    convert_before
        .env("HOME", &home)
        .env("SUPERCODE_HOME", &supercode_home)
        .args(["convert", &runtime_id, "--to", "claude-code", "--out"])
        .arg(&export_before);
    let converted = output_bounded(&mut convert_before, Duration::from_secs(10)).unwrap();
    assert!(converted.status.success(), "{:?}", converted.stderr);
    let exported_before = fs::read(&export_before).unwrap();
    assert!(
        exported_before.starts_with(source_bytes.as_bytes()),
        "same-origin export changed the imported Claude prefix: {}",
        String::from_utf8_lossy(&exported_before)
    );
    let exported_session =
        supercode::Session::from_claude_code_str(std::str::from_utf8(&exported_before).unwrap())
            .unwrap();
    let exported_wire = serde_json::to_string(&exported_session.messages).unwrap();
    for marker in [
        "INTERCHANGE_IMPORTED_FACT",
        "REGISTRY_CODEX_BEFORE_PI",
        "REGISTRY REPLY 0",
        "REGISTRY_PI_MIDDLE",
        "REGISTRY REPLY 1",
    ] {
        assert_eq!(exported_wire.matches(marker).count(), 1, "{exported_wire}");
    }

    // Stock Codex rejoins last and displays both turns. A read-only frontend
    // swap must leave canonical history, persisted bytes, and the eventual
    // Claude-native export byte-identical.
    let final_codex_attach = format!(
        "PATH={}:$PATH HOME={} SUPERCODE_HOME={} TERM=xterm-256color {} attach {} --ui codex --take-control; echo SC_CODEX_FINAL_DONE:$?",
        shell_word(codex_bin.to_str().unwrap()),
        shell_word(home.to_str().unwrap()),
        shell_word(supercode_home.to_str().unwrap()),
        shell_word(bin()),
        shell_word(&runtime_id),
    );
    tmux.send_line(&final_codex_attach);
    tmux.expect("frontend availability: id=codex");
    tmux.expect("REGISTRY REPLY 0");
    tmux.expect("REGISTRY REPLY 1");
    let final_attach_pid = only_child_pid(tmux.pane_pid());
    let final_stock_pid = only_child_pid(final_attach_pid as u32);
    unsafe { libc::kill(-final_stock_pid, libc::SIGKILL) };
    tmux.expect("SC_CODEX_FINAL_DONE:");
    tmux.send_line("echo SC_INTERCHANGE_SHELL_3");
    tmux.expect_count("SC_INTERCHANGE_SHELL_3", 2);
    std::thread::sleep(Duration::from_millis(100));
    assert_eq!(tmux.termios(), original_termios);
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 1);
    assert_eq!(
        http_rpc(
            address,
            token,
            66,
            "history",
            serde_json::json!({"limit":1000}),
        )["result"]["messages"],
        history_after_turns
    );
    assert_eq!(fs::read(&persistence).unwrap(), persisted_after_turns);
    assert_eq!(fs::read_to_string(&source).unwrap(), source_bytes);

    let export_after = home.join("after-final-codex.jsonl");
    let mut convert_after = Command::new(bin());
    convert_after
        .env("HOME", &home)
        .env("SUPERCODE_HOME", &supercode_home)
        .args(["convert", &runtime_id, "--to", "claude-code", "--out"])
        .arg(&export_after);
    let converted = output_bounded(&mut convert_after, Duration::from_secs(10)).unwrap();
    assert!(converted.status.success(), "{:?}", converted.stderr);
    assert_eq!(fs::read(&export_after).unwrap(), exported_before);

    shutdown(address, token);
    assert!(
        wait_child_bounded(&mut host, Duration::from_secs(10), "interchange runtime").success()
    );
    join_thread_bounded(
        provider_task,
        Duration::from_secs(10),
        "interchange provider",
    );
    fs::remove_dir_all(home).ok();
}

/// Hosted proof for the exact stock-client swap required by SUP-61. The
/// release/compatibility job has the same pinned-client prerequisite as the
/// cross-surface stock Codex gate; ordinary workspace tests stay hermetic.
#[cfg(unix)]
#[test]
#[ignore = "requires unmodified codex-cli 0.144.4 and tmux-capable PTYs"]
fn stock_codex_observer_and_embedded_replacement_share_one_runtime() {
    let version = Command::new("codex").arg("--version").output().unwrap();
    assert!(version.status.success());
    assert_eq!(
        String::from_utf8_lossy(&version.stdout).trim(),
        "codex-cli 0.144.4"
    );
    let home = temp_home();
    let supercode_home = home.join("supercode-home");
    fs::create_dir_all(&supercode_home).unwrap();
    fs::write(
        supercode_home.join("config.toml"),
        "approval = \"never\"\nsandbox = \"danger-full-access\"\n\
         [core.retry]\nenabled = false\n\
         [capabilities.server]\nenabled = true\n",
    )
    .unwrap();
    let source = home.join("stock-source.jsonl");
    let source_bytes = serde_json::json!({
        "type": "user",
        "sessionId": "stock-source",
        "cwd": home,
        "timestamp": "2020-01-01T00:00:00Z",
        "message": {"role": "user", "content": "stock imported fact"}
    })
    .to_string()
        + "\n";
    fs::write(&source, &source_bytes).unwrap();
    let (provider, provider_task) = spawn_echo_provider(1);
    let token = "stock-host-token";
    let (mut host, address, _stderr) =
        spawn_registry_host_with_ttl(&home, &supercode_home, &source, provider, token, 30_000);
    let live = list_live_sessions(&home, &supercode_home);
    assert_eq!(live.len(), 1);
    let runtime_id = live[0]["runtime_id"].as_str().unwrap().to_string();
    let persistence = PathBuf::from(live[0]["persistence_location"].as_str().unwrap());
    assert!(unauthorized_http_status(address).contains("401"));

    let tmux = TmuxServer(format!("supercode-sup61-stock-{}", std::process::id()));
    let created = tmux.command(&[
        "new-session",
        "-d",
        "-s",
        "frontend",
        "-x",
        "120",
        "-y",
        "40",
    ]);
    assert!(created.status.success(), "{:?}", created.stderr);
    tmux.send_line("echo SC_SHELL_READY");
    tmux.expect_count("SC_SHELL_READY", 2);
    std::thread::sleep(Duration::from_millis(100));
    let original_termios = tmux.termios();
    let attach_command = format!(
        "HOME={} SUPERCODE_HOME={} TERM=xterm-256color {} attach {} --ui codex --take-control; echo SC_CODEX_DONE:$?",
        shell_word(home.to_str().unwrap()),
        shell_word(supercode_home.to_str().unwrap()),
        shell_word(bin()),
        shell_word(&runtime_id),
    );
    tmux.send_line(&attach_command);
    tmux.expect("frontend availability: id=codex");
    std::thread::sleep(Duration::from_millis(500));
    tmux.send_line("REGISTRY_STOCK_UI");
    tmux.expect("REGISTRY REPLY 0");

    // A second process has observer-only authority and sees the same
    // canonical history while stock Codex owns control.
    let observer_pane = Pty::open(100, 30);
    let observer_before = observer_pane.termios();
    let mut observer = spawn_pty_ui_attach(
        &observer_pane,
        &home,
        &supercode_home,
        &runtime_id,
        "embedded",
        "--observe",
    );
    let mut observer_output = String::new();
    observer_pane.expect(&mut observer_output, "REGISTRY REPLY 0");
    unsafe { libc::kill(observer.id() as libc::pid_t, libc::SIGTERM) };
    assert!(!wait_child_bounded(&mut observer, Duration::from_secs(10), "observer drop").success());
    assert_eq!(observer_pane.termios(), observer_before);
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 1);

    let history_after_turn = http_rpc(
        address,
        token,
        40,
        "history",
        serde_json::json!({"limit":1000}),
    )["result"]["messages"]
        .clone();
    let persistence_after_turn = fs::read(&persistence).unwrap();

    // Crash the isolated stock-client process group. The Supercode host sees
    // the child exit, restores the original pane, detaches its lease, and
    // deliberately leaves the SDK runtime registered.
    let attach_pid = only_child_pid(tmux.pane_pid());
    let stock_pid = only_child_pid(attach_pid as u32);
    unsafe { libc::kill(-stock_pid, libc::SIGKILL) };
    tmux.expect("SC_CODEX_DONE:");
    tmux.send_line("echo SC_AFTER_CRASH_READY");
    tmux.expect_count("SC_AFTER_CRASH_READY", 2);
    std::thread::sleep(Duration::from_millis(100));
    assert_eq!(tmux.termios(), original_termios);
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 1);

    // Replace stock Codex in the exact same tmux pane with the embedded client.
    // Reading and detaching cannot mutate canonical/persisted history.
    let replacement_command = format!(
        "HOME={} SUPERCODE_HOME={} TERM=xterm-256color {} attach {} --ui embedded --take-control; echo SC_EMBEDDED_DONE:$?",
        shell_word(home.to_str().unwrap()),
        shell_word(supercode_home.to_str().unwrap()),
        shell_word(bin()),
        shell_word(&runtime_id),
    );
    tmux.send_line(&replacement_command);
    tmux.expect_count("REGISTRY REPLY 0", 2);
    let eof = tmux.command(&["send-keys", "-t", "frontend", "C-d"]);
    assert!(eof.status.success(), "{:?}", eof.stderr);
    tmux.expect("SC_EMBEDDED_DONE:0");
    tmux.send_line("echo SC_AFTER_REPLACEMENT_READY");
    tmux.expect_count("SC_AFTER_REPLACEMENT_READY", 2);
    std::thread::sleep(Duration::from_millis(100));
    assert_eq!(tmux.termios(), original_termios);
    assert_eq!(
        http_rpc(
            address,
            token,
            41,
            "history",
            serde_json::json!({"limit":1000}),
        )["result"]["messages"],
        history_after_turn
    );
    assert_eq!(fs::read(&persistence).unwrap(), persistence_after_turn);
    assert_eq!(fs::read_to_string(&source).unwrap(), source_bytes);

    shutdown(address, token);
    assert!(wait_child_bounded(&mut host, Duration::from_secs(10), "stock runtime").success());
    join_thread_bounded(provider_task, Duration::from_secs(10), "stock provider");
    fs::remove_dir_all(home).ok();
}

/// Hosted proof for SUP-63's exact stock-Goose boundary. The pinned,
/// unmodified TypeScript/Ink frontend creates only a display session: both
/// Goose launches and the replacement embedded client drive one imported,
/// SDK-owned runtime and one canonical persistence family.
#[cfg(unix)]
#[test]
#[ignore = "requires pinned Goose acd3c135 source, Node 23.9.0, and tmux"]
fn stock_goose_permission_reconnect_and_embedded_replacement_share_one_runtime() {
    let goose_source = std::env::var("SUPERCODE_GOOSE_TUI_SOURCE")
        .expect("set SUPERCODE_GOOSE_TUI_SOURCE to pinned Goose ui/text/src/tui.tsx");
    assert!(goose_source.ends_with("/ui/text/src/tui.tsx"));
    let node = Command::new("node").arg("--version").output().unwrap();
    assert!(node.status.success());
    assert_eq!(String::from_utf8_lossy(&node.stdout).trim(), "v23.9.0");
    let tmux_version = Command::new("tmux").arg("-V").output().unwrap();
    assert!(tmux_version.status.success());

    let home = temp_home();
    let supercode_home = home.join("supercode-home");
    fs::create_dir_all(&supercode_home).unwrap();
    fs::write(
        supercode_home.join("config.toml"),
        "approval = \"on-request\"\nsandbox = \"danger-full-access\"\n\
         [core.retry]\nenabled = false\n\
         [experimental]\nmodule_registry = true\n\
         [capabilities.server]\nenabled = true\n\
         [capabilities.tui]\nenabled = true\n\
         [capabilities.permissions]\nenabled = true\n",
    )
    .unwrap();
    let source = home.join("goose-source.jsonl");
    let source_bytes = serde_json::json!({
        "type": "user",
        "sessionId": "goose-source",
        "cwd": home,
        "timestamp": "2020-01-01T00:00:00Z",
        "message": {"role": "user", "content": "GOOSE_IMPORTED_FACT"}
    })
    .to_string()
        + "\n";
    fs::write(&source, &source_bytes).unwrap();

    let (provider, provider_task) = spawn_goose_proof_provider();
    let token = "goose-proof-token";
    let (mut host, address, _stderr) = spawn_registry_host_with_ttl_and_iterations(
        &home,
        &supercode_home,
        &source,
        provider,
        token,
        30_000,
        3,
    );
    let live = list_live_sessions(&home, &supercode_home);
    assert_eq!(live.len(), 1, "{live:#?}");
    let runtime_id = live[0]["runtime_id"].as_str().unwrap().to_string();
    let persistence = PathBuf::from(live[0]["persistence_location"].as_str().unwrap());

    let tmux = TmuxServer(format!("supercode-sup63-goose-{}", std::process::id()));
    let created = tmux.command(&[
        "new-session",
        "-d",
        "-s",
        "frontend",
        "-x",
        "120",
        "-y",
        "40",
    ]);
    assert!(created.status.success(), "{:?}", created.stderr);
    tmux.send_line("echo SC_GOOSE_SHELL_READY");
    tmux.expect_count("SC_GOOSE_SHELL_READY", 2);

    let goose_attach = |start: &str, done: &str| {
        format!(
            "echo {start}; HOME={} SUPERCODE_HOME={} SUPERCODE_GOOSE_TUI_SOURCE={} TERM=xterm-256color {} attach {} --ui goose --take-control; echo {done}:$?",
            shell_word(home.to_str().unwrap()),
            shell_word(supercode_home.to_str().unwrap()),
            shell_word(&goose_source),
            shell_word(bin()),
            shell_word(&runtime_id),
        )
    };

    tmux.send_line(&goose_attach("SC_GOOSE_ONE_START", "SC_GOOSE_ONE_DONE"));
    // Ink owns the alternate screen and legitimately hides the preceding
    // shell marker. Assert client-specific visible state and unique replies.
    tmux.expect("ready");
    tmux.send_tui_line("GOOSE_PERMISSION_PROMPT");
    tmux.expect("GOOSE_FIRST_DONE");
    let stop = tmux.command(&["send-keys", "-t", "frontend", "C-c"]);
    assert!(stop.status.success(), "{:?}", stop.stderr);
    tmux.expect("SC_GOOSE_ONE_DONE:0");

    let first_history = http_rpc(
        address,
        token,
        40,
        "history",
        serde_json::json!({"limit":1000}),
    )["result"]["messages"]
        .clone();
    let first_wire = first_history.to_string();
    for (marker, expected) in [
        ("GOOSE_IMPORTED_FACT", 1),
        ("GOOSE_PERMISSION_PROMPT", 1),
        // The reversible canonical transcript retains both the exact tool
        // argument and its stdout; either disappearing would be loss.
        ("GOOSE_TOOL_OK", 2),
        ("GOOSE_FIRST_DONE", 1),
    ] {
        assert_eq!(first_wire.matches(marker).count(), expected, "{first_wire}");
    }
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 1);

    // Remove the first alternate-screen residue so `ready` below can only
    // be emitted by the newly connected stock process, never replayed pane
    // text from the detached one.
    tmux.clear_screen_and_history();
    tmux.send_line(&goose_attach("SC_GOOSE_TWO_START", "SC_GOOSE_TWO_DONE"));
    tmux.expect("ready");
    tmux.send_tui_line("GOOSE_RECONNECT_PROMPT");
    tmux.expect("GOOSE_RECONNECT_DONE");
    let stop = tmux.command(&["send-keys", "-t", "frontend", "C-c"]);
    assert!(stop.status.success(), "{:?}", stop.stderr);
    tmux.expect("SC_GOOSE_TWO_DONE:0");
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 1);

    // Likewise force the replacement assertion to come from canonical
    // embedded replay rather than the departed Goose screen.
    tmux.clear_screen_and_history();
    let embedded = format!(
        "echo SC_EMBEDDED_START; HOME={} SUPERCODE_HOME={} TERM=xterm-256color {} attach {} --ui embedded --take-control; echo SC_EMBEDDED_DONE:$?",
        shell_word(home.to_str().unwrap()),
        shell_word(supercode_home.to_str().unwrap()),
        shell_word(bin()),
        shell_word(&runtime_id),
    );
    tmux.send_line(&embedded);
    tmux.expect("GOOSE_RECONNECT_DONE");
    tmux.expect("● Ready");
    tmux.send_line("EMBEDDED_REPLACEMENT_PROMPT");
    tmux.expect("EMBEDDED_REPLACEMENT_DONE");
    tmux.send_line("/detach");
    tmux.expect("SC_EMBEDDED_DONE:0");

    join_thread_bounded(
        provider_task,
        Duration::from_secs(10),
        "Goose proof provider",
    );
    let final_history = http_rpc(
        address,
        token,
        41,
        "history",
        serde_json::json!({"limit":1000}),
    )["result"]["messages"]
        .clone();
    let final_wire = final_history.to_string();
    for (marker, expected) in [
        ("GOOSE_IMPORTED_FACT", 1),
        ("GOOSE_PERMISSION_PROMPT", 1),
        ("GOOSE_TOOL_OK", 2),
        ("GOOSE_FIRST_DONE", 1),
        ("GOOSE_RECONNECT_PROMPT", 1),
        ("GOOSE_RECONNECT_DONE", 1),
        ("EMBEDDED_REPLACEMENT_PROMPT", 1),
        ("EMBEDDED_REPLACEMENT_DONE", 1),
    ] {
        assert_eq!(final_wire.matches(marker).count(), expected, "{final_wire}");
    }
    let persisted = fs::read_to_string(&persistence).unwrap();
    let persisted_messages = persisted
        .lines()
        .filter(|line| !line.trim().is_empty())
        .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
        .collect::<Vec<_>>();
    assert_eq!(serde_json::Value::Array(persisted_messages), final_history);
    assert_eq!(fs::read_to_string(&source).unwrap(), source_bytes);
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 1);

    shutdown(address, token);
    assert!(wait_child_bounded(&mut host, Duration::from_secs(10), "Goose runtime").success());
    assert!(list_live_sessions(&home, &supercode_home).is_empty());
    fs::remove_dir_all(home).ok();
}

/// Remote-boundary proof for SUP-61. Supercode deliberately does not create
/// or own this tunnel: the test starts a stock OpenSSH client exactly as an
/// operator would, then proves auth, drop, expiry, and cursor-safe reattach.
#[cfg(unix)]
#[test]
#[ignore = "requires local OpenSSH sshd and ssh binaries"]
fn operator_created_ssh_tunnel_reconnects_without_stopping_or_replaying_runtime() {
    assert!(std::path::Path::new("/usr/sbin/sshd").is_file());
    let home = temp_home();
    let supercode_home = home.join("supercode-home");
    fs::create_dir_all(&supercode_home).unwrap();
    fs::write(
        supercode_home.join("config.toml"),
        "approval = \"never\"\nsandbox = \"danger-full-access\"\n\
         [core.retry]\nenabled = false\n\
         [capabilities.server]\nenabled = true\n\
         [capabilities.tui]\nenabled = false\n",
    )
    .unwrap();
    let source = home.join("ssh-source.jsonl");
    let source_bytes = serde_json::json!({
        "type": "user",
        "sessionId": "ssh-source",
        "cwd": home,
        "timestamp": "2020-01-01T00:00:00Z",
        "message": {"role": "user", "content": "ssh imported fact"}
    })
    .to_string()
        + "\n";
    fs::write(&source, &source_bytes).unwrap();
    let (provider, provider_task) = spawn_echo_provider(2);
    let bootstrap = "sup61-ssh-bootstrap-secret";
    let (mut host, runtime_address, mut host_stderr) =
        spawn_registry_host_with_ttl(&home, &supercode_home, &source, provider, bootstrap, 5_000);
    let listed = list_live_sessions(&home, &supercode_home);
    assert_eq!(listed.len(), 1);
    let runtime_id = listed[0]["id"].as_str().unwrap();
    let persistence = PathBuf::from(listed[0]["persistence_location"].as_str().unwrap());
    let mut credential_host = Command::new(bin())
        .env("HOME", &home)
        .env("SUPERCODE_HOME", &supercode_home)
        .env_remove("SUPERCODE_SERVER_TOKEN")
        .args(["--quiet", "attach", runtime_id, "--remote-instructions"])
        .stdin(Stdio::piped())
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let credential_host_stdin = credential_host.stdin.take().unwrap();
    let credential_host_stderr = credential_host.stderr.take().unwrap();
    set_nonblocking(&credential_host_stderr);
    let mut credential_host_stderr = BufReader::new(credential_host_stderr);
    let mut credential_diagnostics = String::new();
    let deadline = Instant::now() + Duration::from_secs(10);
    let (credential_path, client_id) = loop {
        let mut line = String::new();
        match credential_host_stderr.read_line(&mut line) {
            Ok(0) => {}
            Ok(_) => credential_diagnostics.push_str(&line),
            Err(error)
                if matches!(
                    error.kind(),
                    std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted
                ) => {}
            Err(error) => panic!("remote recipe stderr failed: {error}"),
        }
        let path = credential_diagnostics
            .lines()
            .find_map(|line| line.split("'cat ").nth(1)?.strip_suffix("')"))
            .map(PathBuf::from);
        let id = credential_diagnostics
            .lines()
            .find_map(|line| line.split("--client-id ").nth(1))
            .map(str::trim)
            .map(str::to_owned);
        if let (Some(path), Some(id)) = (path, id) {
            break (path, id);
        }
        assert!(
            credential_host.try_wait().unwrap().is_none(),
            "credential host exited early: {credential_diagnostics}"
        );
        assert!(Instant::now() < deadline, "remote recipe timed out");
        std::thread::sleep(Duration::from_millis(10));
    };
    let scoped = fs::read_to_string(&credential_path).unwrap();
    assert_eq!(scoped.len(), 64);
    assert_ne!(scoped, bootstrap);
    assert!(!credential_diagnostics.contains(&scoped));
    let ssh = OpenSshFixture::start(home.join("operator-openssh"));

    let (first_tunnel, first_address) = ssh.tunnel(runtime_address);
    assert!(unauthorized_http_status(first_address).contains("401"));
    let first = run_tunneled_frontend(
        &home,
        &supercode_home,
        first_address,
        TunneledCredential {
            ssh: &ssh,
            path: &credential_path,
            token: &scoped,
            client_id: &client_id,
        },
        "REGISTRY_SSH_TUNNEL",
        "REGISTRY REPLY 0",
        None,
    );
    assert!(first.status.success(), "{:?}", first.stderr);
    assert!(!String::from_utf8_lossy(&first.stderr).contains(&scoped));
    let first_rendered = String::from_utf8(first.stdout).unwrap();
    assert!(
        first_rendered.contains("REGISTRY REPLY 0"),
        "{first_rendered}"
    );
    assert!(!first_rendered.contains(&scoped));
    let before_drop = http_rpc_as(
        first_address,
        &scoped,
        Some(&client_id),
        801,
        "frontend.attach",
        serde_json::json!({"limit":1_000}),
    );
    let acknowledged_cursor = before_drop["result"]["history_cursor"].as_u64().unwrap();
    let wrong_client = http_rpc_as(
        first_address,
        &scoped,
        Some("wrong-client"),
        800,
        "frontend.describe",
        serde_json::json!({}),
    );
    assert_eq!(wrong_client["error"]["name"], "unauthorized");
    assert_eq!(wrong_client["error"]["permission"], "client_id");
    let close = http_rpc_as(
        first_address,
        &scoped,
        Some(&client_id),
        799,
        "frontend.close",
        serde_json::json!({}),
    );
    assert_eq!(close["error"]["name"], "unauthorized");
    assert_eq!(close["error"]["permission"], "terminate");
    assert_eq!(
        http_rpc_as(
            first_address,
            &scoped,
            Some(&client_id),
            802,
            "frontend.take_control",
            serde_json::json!({}),
        )["result"]["controller"]["client_id"],
        client_id
    );

    // The network path disappears while the controller lease is held. The
    // runtime is a separate local process and remains live throughout.
    drop(first_tunnel);
    std::thread::sleep(Duration::from_millis(5_500));
    assert!(host.try_wait().unwrap().is_none());
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 1);
    assert_eq!(fs::read_to_string(&source).unwrap(), source_bytes);

    let (_second_tunnel, second_address) = ssh.tunnel(runtime_address);
    assert!(unauthorized_http_status(second_address).contains("401"));
    let stale = http_rpc_as(
        second_address,
        &scoped,
        Some(&client_id),
        803,
        "submit",
        serde_json::json!({"prompt":"REGISTRY_EXPIRED_REMOTE_MUST_NOT_RUN"}),
    );
    assert_eq!(stale["error"]["name"], "lease_expired", "{stale:#}");
    let reclaimed = http_rpc_as(
        second_address,
        &scoped,
        Some(&client_id),
        807,
        "frontend.take_control",
        serde_json::json!({}),
    );
    assert_eq!(reclaimed["result"]["controller"]["client_id"], client_id);

    let second = run_tunneled_frontend(
        &home,
        &supercode_home,
        second_address,
        TunneledCredential {
            ssh: &ssh,
            path: &credential_path,
            token: &scoped,
            client_id: &client_id,
        },
        "REGISTRY_SSH_AFTER_DROP",
        "REGISTRY REPLY 1",
        Some(acknowledged_cursor),
    );
    assert!(second.status.success(), "{:?}", second.stderr);
    assert!(!String::from_utf8_lossy(&second.stderr).contains(&scoped));
    let second_rendered = String::from_utf8(second.stdout).unwrap();
    assert!(
        second_rendered.contains("REGISTRY REPLY 1"),
        "{second_rendered}"
    );
    assert!(!second_rendered.contains(&scoped));
    assert!(
        !second_rendered.contains("REGISTRY REPLY 0"),
        "acknowledged history replayed after reconnect: {second_rendered}"
    );
    let after_reconnect = http_rpc_as(
        second_address,
        &scoped,
        Some(&client_id),
        804,
        "frontend.attach",
        serde_json::json!({"limit":1_000}),
    );
    assert!(
        after_reconnect["result"]["history_cursor"]
            .as_u64()
            .unwrap()
            > acknowledged_cursor
    );
    let history = http_rpc_as(
        second_address,
        &scoped,
        Some(&client_id),
        805,
        "history",
        serde_json::json!({"limit":1_000}),
    )["result"]["messages"]
        .to_string();
    assert_eq!(history.matches("REGISTRY_SSH_TUNNEL").count(), 1);
    assert_eq!(history.matches("REGISTRY_SSH_AFTER_DROP").count(), 1);
    assert_eq!(history.matches("REGISTRY REPLY 0").count(), 1);
    assert_eq!(history.matches("REGISTRY REPLY 1").count(), 1);
    assert!(!history.contains("REGISTRY_EXPIRED_REMOTE_MUST_NOT_RUN"));
    let persisted = fs::read_to_string(&persistence).unwrap();
    assert_eq!(persisted.matches("REGISTRY_SSH_TUNNEL").count(), 1);
    assert_eq!(persisted.matches("REGISTRY_SSH_AFTER_DROP").count(), 1);
    assert_eq!(fs::read_to_string(&source).unwrap(), source_bytes);

    drop(credential_host_stdin);
    assert!(wait_child_bounded(
        &mut credential_host,
        Duration::from_secs(10),
        "remote credential host"
    )
    .success());
    read_to_string_bounded(
        &mut credential_host_stderr,
        &mut credential_diagnostics,
        Duration::from_secs(2),
        "remote credential host diagnostics",
    );
    assert!(!credential_diagnostics.contains(&scoped));
    assert!(!credential_path.exists());
    let revoked = http_rpc_as(
        second_address,
        &scoped,
        Some(&client_id),
        806,
        "frontend.describe",
        serde_json::json!({}),
    );
    assert_eq!(revoked["error"]["name"], "unauthenticated");
    assert!(host.try_wait().unwrap().is_none());

    shutdown(runtime_address, bootstrap);
    assert!(wait_child_bounded(&mut host, Duration::from_secs(10), "SSH runtime").success());
    let mut host_diagnostics = String::new();
    read_to_string_bounded(
        &mut host_stderr,
        &mut host_diagnostics,
        Duration::from_secs(2),
        "SSH runtime diagnostics",
    );
    assert!(
        !host_diagnostics.contains(&scoped) && !host_diagnostics.contains(bootstrap),
        "runtime diagnostics leaked a credential: {host_diagnostics}"
    );
    join_thread_bounded(provider_task, Duration::from_secs(10), "SSH provider");
    fs::remove_dir_all(home).ok();
}

#[cfg(unix)]
#[test]
fn resumed_http_runtime_accepts_a_separate_acp_attachment_without_duplication() {
    let home = temp_home();
    let supercode_home = home.join("supercode-home");
    fs::create_dir_all(&supercode_home).unwrap();
    fs::write(
        supercode_home.join("config.toml"),
        "approval = \"on-request\"\nsandbox = \"danger-full-access\"\n\
         [core.retry]\nenabled = false\n\
         [experimental]\nmodule_registry = true\n\
         [capabilities.server]\nenabled = true\n\
         [capabilities.tui]\nenabled = true\n\
         [capabilities.permissions]\nenabled = true\n\
         ",
    )
    .unwrap();
    let source = home.join("source.jsonl");
    fs::write(
        &source,
        serde_json::json!({
            "type": "user", "sessionId": "attach-source", "cwd": home,
            "timestamp": "2020-01-01T00:00:00Z",
            "message": {"role": "user", "content": "imported context"}
        })
        .to_string()
            + "\n",
    )
    .unwrap();
    let source_before = fs::read(&source).unwrap();
    let mut tmux_version = Command::new("tmux");
    tmux_version.arg("-V");
    let tmux_available = output_bounded(&mut tmux_version, Duration::from_secs(3))
        .is_ok_and(|output| output.status.success());
    let (provider, provider_task, interrupt_started, release_first_turn) =
        spawn_provider(tmux_available);
    let token = "attach-test-token";
    let mut host = Command::new(bin())
        .env("HOME", &home)
        .env("SUPERCODE_HOME", &supercode_home)
        .env_remove("OPENROUTER_API_KEY")
        .args([
            "--quiet",
            "--no-reduced",
            "--api-key",
            "x",
            "--base-url",
            &format!("http://{provider}"),
            "--max-iterations",
            "3",
            "resume",
            source.to_str().unwrap(),
            "--paused",
            "--serve",
            "--no-tmux",
            "--bind",
            "127.0.0.1:0",
            "--token",
            token,
        ])
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let host_stderr = host.stderr.take().unwrap();
    set_nonblocking(&host_stderr);
    let mut host_stderr = BufReader::new(host_stderr);
    let address_deadline = Instant::now() + Duration::from_secs(15);
    let address = loop {
        assert!(
            Instant::now() < address_deadline,
            "host did not publish its address within 15 seconds"
        );
        let mut line = String::new();
        match host_stderr.read_line(&mut line) {
            Ok(0) => panic!("host exited before publishing its address"),
            Ok(_) => {
                if let Some(value) = line.split("listening on http://").nth(1) {
                    break value.trim().parse::<SocketAddr>().unwrap();
                }
            }
            Err(error)
                if error.kind() == std::io::ErrorKind::WouldBlock
                    || error.kind() == std::io::ErrorKind::Interrupted =>
            {
                std::thread::sleep(Duration::from_millis(10));
            }
            Err(error) => panic!("host stderr read failed: {error}"),
        }
    };

    let mut live_list = Command::new(bin());
    live_list
        .env("HOME", &home)
        .env("SUPERCODE_HOME", &supercode_home)
        .args(["sessions", "list", "--live", "--json"]);
    let live_list = output_bounded(&mut live_list, Duration::from_secs(10)).unwrap();
    assert!(live_list.status.success(), "{:?}", live_list.stderr);
    let live_entries: serde_json::Value = serde_json::from_slice(&live_list.stdout).unwrap();
    let live_entry = &live_entries.as_array().unwrap()[0];
    assert_eq!(live_entry["source_harness"], "claude-code");
    assert_eq!(live_entry["source_session_id"], "attach-source");
    assert_eq!(live_entry["state"], "idle");
    assert_eq!(
        live_entry["endpoint_capabilities"],
        serde_json::json!(["http", "acp"])
    );
    assert_eq!(live_entry["actions"]["submit"], false);
    assert_eq!(live_entry["observers"], serde_json::json!([]));

    let mut bridge = Command::new(bin())
        .env("HOME", &home)
        .env("SUPERCODE_HOME", &supercode_home)
        .args([
            "--quiet",
            "acp",
            "--connect",
            &format!("http://{address}"),
            "--token",
            token,
        ])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let mut bridge_stdin = bridge.stdin.take().unwrap();
    let bridge_stdout = bridge.stdout.take().unwrap();
    set_nonblocking(&bridge_stdout);
    let mut bridge_stdout = BufReader::new(bridge_stdout);
    send(
        &mut bridge_stdin,
        serde_json::json!({
            "jsonrpc": "2.0", "id": 1, "method": "initialize",
            "params": {"protocolVersion": 1, "clientCapabilities": {}}
        }),
    );
    assert_eq!(receive(&mut bridge_stdout)["result"]["protocolVersion"], 1);
    send(
        &mut bridge_stdin,
        serde_json::json!({
            "jsonrpc": "2.0", "id": 2, "method": "session/new",
            "params": {"cwd": home, "mcpServers": []}
        }),
    );
    let session_id = receive(&mut bridge_stdout)["result"]["sessionId"]
        .as_str()
        .unwrap()
        .to_string();
    let http_snapshot = http_rpc(
        address,
        token,
        19,
        "frontend.attach",
        serde_json::json!({"limit":1_000}),
    );
    send(
        &mut bridge_stdin,
        serde_json::json!({
            "jsonrpc":"2.0", "id":20,
            "method":"supercode/frontend/attach",
            "params":{"sessionId":session_id,"limit":1_000,"afterSequence":0}
        }),
    );
    let acp_snapshot = receive(&mut bridge_stdout);
    assert_eq!(
        acp_snapshot["result"]["history"], http_snapshot["result"]["history"],
        "ACP and HTTP must expose byte-identical canonical history"
    );
    assert_eq!(
        acp_snapshot["result"]["history_cursor"],
        http_snapshot["result"]["history_cursor"]
    );
    send(
        &mut bridge_stdin,
        serde_json::json!({
            "jsonrpc": "2.0", "id": 3, "method": "session/prompt",
            "params": {
                "sessionId": session_id,
                "prompt": [{"type": "text", "text": "ATTACHED_ACP_PROMPT"}]
            }
        }),
    );
    let mut messages = Vec::new();
    loop {
        let message = receive(&mut bridge_stdout);
        let first_chunk = message
            .pointer("/params/event/payload/text")
            .and_then(serde_json::Value::as_str)
            == Some("SHARED ");
        messages.push(message);
        if first_chunk {
            break;
        }
    }
    assert!(messages.iter().any(|message| {
        message
            .pointer("/params/event/payload/text")
            .and_then(serde_json::Value::as_str)
            == Some("SHARED ")
    }));
    let last_acknowledged = messages
        .iter()
        .filter_map(|message| {
            message
                .pointer("/params/event/sequence")
                .and_then(serde_json::Value::as_u64)
        })
        .max()
        .expect("ACP bridge received canonical event notifications");
    // Disconnect while the provider still holds the second chunk. The ACP
    // bridge is non-owning, so the hosted SDK turn must continue.
    drop(bridge_stdin);
    assert!(wait_child_bounded(&mut bridge, Duration::from_secs(10), "ACP bridge").success());

    // The first-class ACP frontend adapter feeds the exact same line/TUI
    // runtime contract as HTTP. It discovers this already-running session,
    // replays canonical history, and detaches without stopping the host.
    let acp_frontend_home = temp_home();
    let acp_frontend_supercode_home = acp_frontend_home.join("supercode-home");
    fs::create_dir_all(&acp_frontend_supercode_home).unwrap();
    fs::write(
        acp_frontend_supercode_home.join("config.toml"),
        "[capabilities.tui]\nenabled = false\n",
    )
    .unwrap();
    let mut acp_frontend = Command::new(bin())
        .env("HOME", &acp_frontend_home)
        .env("SUPERCODE_HOME", &acp_frontend_supercode_home)
        .env("SUPERCODE_SERVER_TOKEN", token)
        .args(["attach", "--acp", "--connect", &format!("http://{address}")])
        .args(["--after-sequence", &last_acknowledged.to_string()])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let mut acp_frontend_stdin = acp_frontend.stdin.take().unwrap();
    let acp_frontend_stdout = acp_frontend.stdout.take().unwrap();
    set_nonblocking(&acp_frontend_stdout);
    let mut acp_frontend_stdout = BufReader::new(acp_frontend_stdout);
    let mut acp_rendered = String::new();
    read_line_until(&mut acp_frontend_stdout, &mut acp_rendered, "attached to");
    release_first_turn
        .send(())
        .expect("provider release channel closed");
    read_line_until(&mut acp_frontend_stdout, &mut acp_rendered, "RUNTIME DONE");
    writeln!(acp_frontend_stdin, "/detach").unwrap();
    drop(acp_frontend_stdin);
    assert!(wait_child_bounded(
        &mut acp_frontend,
        Duration::from_secs(10),
        "ACP frontend detach"
    )
    .success());
    assert!(!acp_rendered.contains("\"payload\":"), "{acp_rendered}");
    assert!(!acp_rendered.contains("\"type\":"), "{acp_rendered}");
    assert!(
        !acp_rendered.contains("SHARED "),
        "the acknowledged first chunk was replayed: {acp_rendered}"
    );
    assert!(host.try_wait().unwrap().is_none());

    let checkpoint_dir = acp_frontend_supercode_home.join("frontend-acp");
    let checkpoint_path = fs::read_dir(&checkpoint_dir)
        .unwrap()
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .find(|path| path.extension().and_then(|extension| extension.to_str()) == Some("json"))
        .expect("the first ACP frontend persisted its delivery checkpoint");
    let checkpoint: supercode::AcpFrontendCheckpoint =
        serde_json::from_slice(&fs::read(&checkpoint_path).unwrap()).unwrap();
    assert_eq!(checkpoint.schema_version, 1);
    assert_eq!(checkpoint.session_id, session_id);
    assert!(checkpoint.acknowledged_sequence > last_acknowledged);

    // Restart the public client without a caller-supplied cursor. It must
    // recover the session-bound checkpoint from disk and avoid rendering any
    // already acknowledged history or streamed chunk.
    let mut restarted_acp = Command::new(bin())
        .env("HOME", &acp_frontend_home)
        .env("SUPERCODE_HOME", &acp_frontend_supercode_home)
        .env("SUPERCODE_SERVER_TOKEN", token)
        .args(["attach", "--acp", "--connect", &format!("http://{address}")])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let mut restarted_stdin = restarted_acp.stdin.take().unwrap();
    let restarted_stdout = restarted_acp.stdout.take().unwrap();
    set_nonblocking(&restarted_stdout);
    let mut restarted_stdout = BufReader::new(restarted_stdout);
    let mut restarted_rendered = String::new();
    read_line_until(
        &mut restarted_stdout,
        &mut restarted_rendered,
        "attached to",
    );
    writeln!(restarted_stdin, "/detach").unwrap();
    drop(restarted_stdin);
    assert!(wait_child_bounded(
        &mut restarted_acp,
        Duration::from_secs(10),
        "checkpoint-restored ACP frontend detach"
    )
    .success());
    read_to_string_bounded(
        &mut restarted_stdout,
        &mut restarted_rendered,
        Duration::from_secs(2),
        "checkpoint-restored ACP stdout",
    );
    assert!(
        !restarted_rendered.contains("SHARED ") && !restarted_rendered.contains("RUNTIME DONE"),
        "the disk-restored ACP frontend duplicated acknowledged content: {restarted_rendered}"
    );
    assert!(host.try_wait().unwrap().is_none());
    fs::remove_dir_all(acp_frontend_home).ok();

    // Drive a complete tool/approval turn through the ACP implementation of
    // SdkRuntime. The stock terminal renderer and composer are unchanged;
    // only the transport object supplied to them differs.
    let acp_pty = Pty::open(72, 20);
    let acp_before = acp_pty.termios();
    let mut acp_terminal = spawn_pty_acp_attach(&acp_pty, &home, address, token);
    let mut acp_terminal_rendered = String::new();
    acp_pty.expect(&mut acp_terminal_rendered, "SHARED");
    acp_pty.paste_and_enter("ACP_FRONTEND_TOOL_PROMPT");
    acp_pty.expect(&mut acp_terminal_rendered, "Approval required");
    let acp_completion_start = acp_terminal_rendered.len();
    acp_pty.write(b"y");
    acp_pty.expect(&mut acp_terminal_rendered, "list_dir");
    acp_pty.expect_after(&mut acp_terminal_rendered, acp_completion_start, "DONE");
    acp_pty.paste_and_enter("/detach");
    wait_success(&mut acp_terminal, &acp_pty, &mut acp_terminal_rendered);
    assert_eq!(acp_pty.termios(), acp_before);
    assert!(
        !acp_terminal_rendered.contains("\"payload\":")
            && !acp_terminal_rendered.contains("\"type\":"),
        "{acp_terminal_rendered}"
    );
    assert!(host.try_wait().unwrap().is_none());
    let acp_http_history = http_rpc(
        address,
        token,
        97,
        "frontend.attach",
        serde_json::json!({"limit":1_000}),
    )["result"]["history"]
        .clone();

    // A real controlling PTY enters the shared full-screen frontend. Its
    // exact termios fields, control-character array, and speeds must survive
    // success and typed detach unchanged.
    let pty = Pty::open(72, 20);
    let before = pty.termios();
    let mut terminal = spawn_pty_attach(&pty, &home, address, token);
    let mut rendered = String::new();
    pty.expect(&mut rendered, "SHARED");
    pty.paste_and_enter("ATTACHED_TERMINAL_PROMPT");
    pty.expect(&mut rendered, "Approval required");
    pty.write(b"y");
    pty.expect(&mut rendered, "list_dir");
    pty.expect(&mut rendered, "TERMINAL");
    pty.paste_and_enter("/detach");
    wait_success(&mut terminal, &pty, &mut rendered);
    assert_eq!(pty.termios(), before, "terminal modes must restore exactly");
    assert!(
        !rendered.contains("\"type\":"),
        "raw event JSON leaked: {rendered}"
    );

    // A second real frontend immediately replays the completed long answer;
    // neither detach shuts down the hosted SDK runtime.
    let reattach_pty = Pty::open(72, 20);
    let reattach_before = reattach_pty.termios();
    let mut reattach = spawn_pty_attach(&reattach_pty, &home, address, token);
    let mut replayed = String::new();
    reattach_pty.expect(&mut replayed, "TERMINAL");
    let composer_start = replayed.len();
    reattach_pty.paste("INTERRUPT_PROMPT");
    // Seeing the newly pasted text proves this frontend has completed its
    // startup cursor exchange and processed the composer event. Submit only
    // after that milestone; replayed history alone is not an input-ready
    // signal for a just-created full-screen terminal.
    reattach_pty.expect_after(&mut replayed, composer_start, "INTERRUPT_PROMPT");
    reattach_pty.write(b"\r");
    let interrupt_target = interrupt_started
        .recv_timeout(Duration::from_secs(10))
        .expect("interrupt target reached provider");
    assert_eq!(interrupt_target, "pty");
    reattach_pty.write(&[0x03]);
    reattach_pty.expect(&mut replayed, "interrupted");
    reattach_pty.paste_and_enter("/detach");
    wait_success(&mut reattach, &reattach_pty, &mut replayed);
    assert_eq!(reattach_pty.termios(), reattach_before);
    assert!(
        !replayed.contains("\"payload\":"),
        "raw event JSON leaked: {replayed}"
    );

    // A built binary on a real controlling PTY must still choose the line
    // frontend when the trusted config explicitly disables the TUI module.
    // This is distinct from redirected-stdio fallback: stdin/stdout/stderr
    // here are the same real PTY used by the full-screen restoration proof.
    let disabled_home = temp_home();
    let disabled_supercode_home = disabled_home.join("supercode-home");
    fs::create_dir_all(&disabled_supercode_home).unwrap();
    fs::write(
        disabled_supercode_home.join("config.toml"),
        "[capabilities.tui]\nenabled = false\n",
    )
    .unwrap();
    let disabled_pty = Pty::open(72, 20);
    let disabled_before = disabled_pty.termios();
    let mut disabled_frontend = spawn_pty_attach(&disabled_pty, &disabled_home, address, token);
    let mut disabled_output = String::new();
    disabled_pty.expect(&mut disabled_output, "attached to");
    disabled_pty.expect(&mut disabled_output, "TERMINAL RUNTIME DONE");
    disabled_pty.write(b"/detach\r");
    wait_success(&mut disabled_frontend, &disabled_pty, &mut disabled_output);
    assert_eq!(disabled_pty.termios(), disabled_before);
    assert!(
        !disabled_output.contains("\u{1b}[?1049h") && !disabled_output.contains("\u{1b}[?2004h"),
        "TUI-disabled PTY entered full-screen terminal modes: {disabled_output:?}"
    );
    assert!(
        !disabled_output.contains("\"type\":") && !disabled_output.contains("\"payload\":"),
        "TUI-disabled PTY leaked runtime JSON: {disabled_output}"
    );
    fs::remove_dir_all(disabled_home).ok();

    // Even with the TUI module enabled, redirected stdio keeps the generic
    // line frontend. Human wording is not byte-stable across the unified
    // frontend change, but content ordering and terminal/JSON safety are.
    let line_home = temp_home();
    let line_supercode_home = line_home.join("supercode-home");
    fs::create_dir_all(&line_supercode_home).unwrap();
    fs::write(
        line_supercode_home.join("config.toml"),
        "[capabilities.tui]\nenabled = true\n",
    )
    .unwrap();
    let mut line = Command::new(bin())
        .env("HOME", &line_home)
        .env("SUPERCODE_HOME", &line_supercode_home)
        .args([
            "attach",
            "--connect",
            &format!("http://{address}"),
            "--token",
            token,
        ])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let mut line_stdin = line.stdin.take().unwrap();
    let line_stdout = line.stdout.take().unwrap();
    // Keep milestone waits bounded. A blocking pipe would otherwise make a
    // failed line-render assertion hang the entire integration test.
    set_nonblocking(&line_stdout);
    let mut line_stdout = BufReader::new(line_stdout);
    let mut line_stderr = line.stderr.take().unwrap();
    set_nonblocking(&line_stderr);
    let mut line_rendered = String::new();

    writeln!(line_stdin, "LINE_TOOL_PROMPT").unwrap();
    line_stdin.flush().unwrap();
    let request_deadline = std::time::Instant::now() + Duration::from_secs(10);
    let request_id = loop {
        let snapshot = http_rpc(
            address,
            token,
            70,
            "frontend.attach",
            serde_json::json!({"limit": 1_000}),
        );
        let replay = snapshot["result"]["replay"].as_array().unwrap();
        let resolved = replay
            .iter()
            .filter(|event| event["kind"] == "request_resolved")
            .filter_map(|event| event["payload"]["request_id"].as_u64())
            .collect::<std::collections::HashSet<_>>();
        if let Some(request_id) = replay.iter().rev().find_map(|event| {
            (event["kind"] == "request")
                .then(|| event["payload"]["request"]["id"].as_u64())
                .flatten()
                .filter(|id| !resolved.contains(id))
        }) {
            break request_id;
        }
        assert!(
            std::time::Instant::now() < request_deadline,
            "line tool approval did not become pending: {snapshot}"
        );
        std::thread::sleep(Duration::from_millis(10));
    };
    let takeover = http_rpc(
        address,
        token,
        70,
        "frontend.take_control",
        serde_json::json!({}),
    );
    assert_eq!(
        takeover["result"]["controller"]["client_id"], "legacy-owner",
        "approval responder must take control explicitly: {takeover}"
    );
    let response = http_rpc(
        address,
        token,
        71,
        "respond",
        serde_json::json!({
            "response": {
                "kind": "approval",
                "request_id": request_id,
                "decision": "allow"
            }
        }),
    );
    assert_eq!(response["result"]["accepted"], true, "{response}");
    let detached = http_rpc(address, token, 72, "frontend.detach", serde_json::json!({}));
    assert!(detached.get("result").is_some(), "{detached}");
    read_line_until(&mut line_stdout, &mut line_rendered, "LINE_TOOL_DONE");
    let line_reply_end = line_rendered.len();
    read_line_until_after(
        &mut line_stdout,
        &mut line_rendered,
        line_reply_end,
        "[event: turn_completed; details unavailable in line mode]",
    );
    std::thread::sleep(Duration::from_millis(50));
    assert!(
        line_rendered.contains("[working: list_dir]"),
        "{line_rendered}"
    );
    assert!(
        line_rendered.contains("[runtime input requested; attach from a TTY to respond]"),
        "{line_rendered}"
    );
    assert!(
        line_rendered.contains("[list_dir: done]"),
        "{line_rendered}"
    );
    assert!(
        line_rendered.contains("LINE_SAFE\nLINE_TOOL_DONE"),
        "{line_rendered}"
    );
    assert!(
        line_rendered.contains("[event: usage; details unavailable in line mode]"),
        "{line_rendered}"
    );
    let live_tool_start = line_rendered.rfind("LINE_SAFE").unwrap();
    let live_tool_rendered = &line_rendered[live_tool_start..];
    assert_eq!(live_tool_rendered.matches("LINE_TOOL_DONE").count(), 1);
    assert_eq!(
        live_tool_rendered
            .matches("[event: usage; details unavailable in line mode]")
            .count(),
        1
    );

    let line_interrupt_start = line_rendered.len();
    writeln!(line_stdin, "LINE_INTERRUPT_PROMPT").unwrap();
    line_stdin.flush().unwrap();
    assert_eq!(
        interrupt_started
            .recv_timeout(Duration::from_secs(10))
            .expect("line interrupt target reached provider"),
        "line"
    );
    writeln!(line_stdin, "/interrupt").unwrap();
    line_stdin.flush().unwrap();
    read_line_until_after(
        &mut line_stdout,
        &mut line_rendered,
        line_interrupt_start,
        "[turn interrupted]",
    );

    let line_error_start = line_rendered.len();
    writeln!(line_stdin, "LINE_ERROR_PROMPT").unwrap();
    line_stdin.flush().unwrap();
    read_line_until_after(
        &mut line_stdout,
        &mut line_rendered,
        line_error_start,
        "[turn failed:",
    );
    writeln!(line_stdin, "/detach").unwrap();
    drop(line_stdin);
    assert!(
        wait_child_bounded(&mut line, Duration::from_secs(10), "line frontend detach").success()
    );
    read_to_string_bounded(
        &mut line_stdout,
        &mut line_rendered,
        Duration::from_secs(2),
        "line stdout",
    );
    let mut line_errors = String::new();
    read_to_string_bounded(
        &mut line_stderr,
        &mut line_errors,
        Duration::from_secs(2),
        "line stderr",
    );
    // The canonical SSE lifecycle can arrive before the submit POST returns,
    // so a successful detach may make its later stderr diagnostic optional.
    assert!(!line_errors.contains('\u{1b}'), "{line_errors:?}");
    assert!(!line_errors.contains("\"type\":"), "{line_errors}");
    assert!(!line_errors.contains("\"payload\":"), "{line_errors}");
    assert!(
        line_rendered.len() <= 64 * 1024,
        "line rendering exceeded its proof bound: {} bytes",
        line_rendered.len()
    );
    assert_eq!(
        line_rendered[line_interrupt_start..line_error_start]
            .matches("[turn interrupted]")
            .count(),
        1,
        "live interrupt was missing or duplicated: {line_rendered}"
    );
    assert_eq!(
        line_rendered[line_error_start..]
            .matches("[turn failed:")
            .count(),
        1,
        "live failure was missing or duplicated: {line_rendered}"
    );
    assert!(line_rendered.contains("attached to"), "{line_rendered}");
    assert!(
        line_rendered.contains("TERMINAL RUNTIME DONE"),
        "{line_rendered}"
    );
    assert!(
        line_rendered.find("TERMINAL RUNTIME DONE") < line_rendered.find("LINE_SAFE"),
        "history must precede the live answer: {line_rendered}"
    );
    assert!(
        !line_rendered.contains("PWNED"),
        "OSC title contents leaked: {line_rendered:?}"
    );
    assert!(
        !line_rendered.contains("PAYLOAD"),
        "unknown-event payload contents leaked: {line_rendered:?}"
    );
    assert!(!line_rendered.contains('\u{1b}'), "{line_rendered:?}");
    assert!(!line_rendered.contains('\u{7}'), "{line_rendered:?}");
    assert!(!line_rendered.contains("\"type\":"), "{line_rendered}");
    assert!(!line_rendered.contains("\"payload\":"), "{line_rendered}");
    fs::remove_dir_all(line_home).ok();

    // Run the same frontend in an actual tmux pane when tmux is available.
    // This is additive to the portable openpty proof above: CI hosts without
    // tmux still verify terminal semantics, while developer machines prove
    // tmux replay, input, approval/respond, interruption, and detach.
    if tmux_available {
        let tmux = TmuxServer(format!("supercode-sup45-{}", std::process::id()));
        let address_arg = format!("http://{address}");
        let mut start_command = Command::new("tmux");
        start_command
            .env("HOME", &home)
            .env("SUPERCODE_HOME", &supercode_home)
            .args([
                "-L",
                &tmux.0,
                "new-session",
                "-d",
                "-x",
                "72",
                "-y",
                "20",
                "-s",
                "frontend",
            ])
            .arg(bin())
            .args(["attach", "--connect", &address_arg, "--token", token]);
        let start = output_bounded(&mut start_command, Duration::from_secs(5)).unwrap();
        assert!(
            start.status.success(),
            "tmux start failed: {}",
            String::from_utf8_lossy(&start.stderr)
        );
        tmux.expect("LINE_ERROR_PROMPT");
        tmux.send_line("TMUX_PROMPT");
        tmux.expect("Approval required");
        let tool_allowed = tmux.command(&["send-keys", "-t", "frontend", "y"]);
        assert!(tool_allowed.status.success(), "{:?}", tool_allowed.stderr);
        tmux.expect("list_dir");
        let tmux_rendered = tmux.expect("TMUX RUNTIME DONE");
        assert!(!tmux_rendered.contains("\"payload\":"), "{tmux_rendered}");
        assert!(!tmux_rendered.contains("\"type\":"), "{tmux_rendered}");
        tmux.send_line("TMUX_INTERRUPT_PROMPT");
        let tmux_interrupt_target = interrupt_started
            .recv_timeout(Duration::from_secs(2))
            .or_else(|_| {
                let enter = tmux.command(&["send-keys", "-t", "frontend", "Enter"]);
                assert!(enter.status.success(), "{:?}", enter.stderr);
                interrupt_started.recv_timeout(Duration::from_secs(10))
            })
            .expect("tmux interrupt target reached provider");
        assert_eq!(tmux_interrupt_target, "tmux");
        let interrupted = tmux.command(&["send-keys", "-t", "frontend", "C-c"]);
        assert!(interrupted.status.success(), "{:?}", interrupted.stderr);
        let tmux_interrupted = tmux.expect_after("TMUX_INTERRUPT_PROMPT", "Turn interrupted");
        assert!(
            !tmux_interrupted.contains("submit_failure"),
            "{tmux_interrupted}"
        );
        tmux.send_line("TMUX_APPROVAL_PROMPT");
        tmux.expect("Approval required");
        let allowed = tmux.command(&["send-keys", "-t", "frontend", "y"]);
        assert!(allowed.status.success(), "{:?}", allowed.stderr);
        tmux.expect("TMUX APPROVAL DONE");
        tmux.send_line("/detach");
        tmux.wait_for_exit();
    }

    // A provider failure is a turn error, not a frontend failure. The live
    // TTY stays usable, renders the semantic error, and restores exactly on
    // the following typed detach.
    let error_pty = Pty::open(72, 20);
    let error_before = error_pty.termios();
    let mut error_frontend = spawn_pty_attach(&error_pty, &home, address, token);
    let mut error_output = String::new();
    error_pty.expect(&mut error_output, "Ready");
    error_pty.paste_and_enter("ERROR_PROMPT");
    error_pty.expect(&mut error_output, "Error:");
    error_pty.paste_and_enter("/detach");
    wait_success(&mut error_frontend, &error_pty, &mut error_output);
    assert_eq!(error_pty.termios(), error_before);

    // Every frontend detach left the hosted runtime alive.
    assert!(host.try_wait().unwrap().is_none());

    // A final live PTY exits through remote SSE disconnect rather than a
    // local key, and still restores every terminal field exactly.
    let disconnect_pty = Pty::open(72, 20);
    let disconnect_before = disconnect_pty.termios();
    let mut disconnect = spawn_pty_attach(&disconnect_pty, &home, address, token);
    let mut disconnected_output = String::new();
    disconnect_pty.expect(&mut disconnected_output, "Ready");
    shutdown(address, token);
    wait_success(&mut disconnect, &disconnect_pty, &mut disconnected_output);
    assert_eq!(disconnect_pty.termios(), disconnect_before);
    assert!(wait_child_bounded(&mut host, Duration::from_secs(10), "host shutdown").success());
    join_thread_bounded(provider_task, Duration::from_secs(2), "provider task");
    assert_eq!(fs::read(&source).unwrap(), source_before);
    let sessions = supercode_home.join("sessions");
    let persisted_path = fs::read_dir(sessions)
        .unwrap()
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .find(|path| {
            let is_working_transcript = path
                .file_name()
                .and_then(|name| name.to_str())
                .is_some_and(|name| name.ends_with(".jsonl") && !name.ends_with(".sidecar.jsonl"));
            if !is_working_transcript {
                return false;
            }
            fs::read_to_string(path)
                .unwrap_or_default()
                .contains("TERMINAL RUNTIME DONE")
        })
        .expect("persisted hosted runtime transcript");
    let persisted_messages = fs::read_to_string(persisted_path)
        .unwrap()
        .lines()
        .filter(|line| !line.trim().is_empty())
        .map(|line| serde_json::from_str::<supercode::ChatMessage>(line).unwrap())
        .collect::<Vec<_>>();
    let http_messages = serde_json::from_value::<Vec<supercode::ChatMessage>>(acp_http_history)
        .expect("typed HTTP history");
    assert!(persisted_messages.len() >= http_messages.len());
    assert_eq!(
        serde_json::to_vec(&persisted_messages[..http_messages.len()]).unwrap(),
        serde_json::to_vec(&http_messages).unwrap(),
        "ACP/HTTP transcript through the ACP approval turn must equal the runtime owner's persisted canonical transcript byte-for-byte"
    );
    fs::remove_dir_all(home).ok();
}

#[cfg(unix)]
#[test]
fn default_tmux_supervision_is_attachable_reusable_and_persistence_safe() {
    let mut tmux_version = Command::new("tmux");
    tmux_version.arg("-V");
    if !output_bounded(&mut tmux_version, Duration::from_secs(3))
        .is_ok_and(|output| output.status.success())
    {
        eprintln!("skipping real tmux proof because tmux is unavailable");
        return;
    }

    let home = temp_home();
    let supercode_home = home.join("supercode-home");
    fs::create_dir_all(&supercode_home).unwrap();
    fs::write(
        supercode_home.join("config.toml"),
        "approval = \"never\"\nsandbox = \"danger-full-access\"\n\
         [core.retry]\nenabled = false\n\
         [experimental]\nmodule_registry = true\n\
         [capabilities.server]\nenabled = true\n\
         [capabilities.tui]\nenabled = true\n\
         [hooks]\n\
         session_start = \"echo session_start >> $HOOKS_LOG\"\n\
         session_end = \"echo session_end >> $HOOKS_LOG\"\n",
    )
    .unwrap();
    let hooks_log = home.join("hooks.log");
    let source_one = home.join("source-one.jsonl");
    let source_two = home.join("source-two.jsonl");
    let write_source = |path: &std::path::Path, id: &str| {
        fs::write(
            path,
            serde_json::json!({
                "type": "user", "sessionId": id, "cwd": home,
                "timestamp": "2020-01-01T00:00:00Z",
                "message": {"role": "user", "content": format!("imported {id}")}
            })
            .to_string()
                + "\n",
        )
        .unwrap();
    };
    write_source(&source_one, "tmux-source-one");
    write_source(&source_two, "tmux-source-two");
    let source_one_before = fs::read(&source_one).unwrap();
    let source_two_before = fs::read(&source_two).unwrap();
    let (provider, provider_task) = spawn_echo_provider(2);

    let start = |source: &std::path::Path| {
        let mut command = Command::new(bin());
        command
            .env("HOME", &home)
            .env("SUPERCODE_HOME", &supercode_home)
            .env("HOOKS_LOG", &hooks_log)
            .env_remove("OPENROUTER_API_KEY")
            .args([
                "--quiet",
                "--cwd",
                home.to_str().unwrap(),
                "--no-reduced",
                "--api-key",
                "private-api-key",
                "--base-url",
                &format!("http://{provider}"),
                "--max-iterations",
                "1",
                "resume",
                source.to_str().unwrap(),
                "--paused",
                "--serve",
                "--bind",
                "127.0.0.1:0",
                "--token",
                "private-runtime-token",
            ]);
        output_bounded(&mut command, Duration::from_secs(25)).unwrap()
    };

    let started_one = start(&source_one);
    assert!(started_one.status.success(), "{:?}", started_one.stderr);
    let started_two = start(&source_two);
    assert!(started_two.status.success(), "{:?}", started_two.stderr);
    let first_stdout = String::from_utf8(started_one.stdout).unwrap();
    let second_stdout = String::from_utf8(started_two.stdout).unwrap();
    assert!(first_stdout.contains("started runtime"), "{first_stdout}");
    assert!(second_stdout.contains("started runtime"), "{second_stdout}");
    assert!(!first_stdout.contains("private-api-key"), "{first_stdout}");
    assert!(
        !first_stdout.contains("private-runtime-token"),
        "{first_stdout}"
    );

    let live = list_live_sessions(&home, &supercode_home);
    assert_eq!(live.len(), 2, "{live:?}");
    assert_eq!(
        fs::read_to_string(&hooks_log).unwrap().lines().count(),
        2,
        "only the two owner processes may fire session_start"
    );
    let find = |source_id: &str| {
        live.iter()
            .find(|entry| entry["source_session_id"] == source_id)
            .unwrap()
    };
    let first = find("tmux-source-one");
    let second = find("tmux-source-two");
    let first_id = first["runtime_id"].as_str().unwrap().to_string();
    let second_id = second["runtime_id"].as_str().unwrap().to_string();
    let first_tmux = first["supervisor"]["session_name"]
        .as_str()
        .unwrap()
        .to_string();
    let second_tmux = second["supervisor"]["session_name"]
        .as_str()
        .unwrap()
        .to_string();
    assert_ne!(first_tmux, second_tmux);
    for tmux_name in [&first_tmux, &second_tmux] {
        let status = Command::new("tmux")
            .args(["has-session", "-t", tmux_name])
            .status()
            .unwrap();
        assert!(status.success(), "missing {tmux_name}");
    }
    let pane_commands = Command::new("tmux")
        .args(["list-panes", "-a", "-F", "#{pane_start_command}"])
        .output()
        .unwrap();
    let pane_commands = String::from_utf8(pane_commands.stdout).unwrap();
    assert!(
        !pane_commands.contains("private-api-key"),
        "{pane_commands}"
    );
    assert!(
        !pane_commands.contains("private-runtime-token"),
        "{pane_commands}"
    );

    // Crashing only the presentation window leaves the SDK owner registered.
    assert!(Command::new("tmux")
        .args(["kill-window", "-t", &format!("{first_tmux}:frontend")])
        .status()
        .unwrap()
        .success());
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 2);

    // A completely separate registry-backed terminal reconnects, drives a
    // turn, and detaches without closing the owner.
    let pty = Pty::open(80, 24);
    let before = pty.termios();
    let mut frontend = spawn_pty_registry_attach(&pty, &home, &supercode_home, &first_id);
    let mut rendered = String::new();
    pty.expect(&mut rendered, "Ready");
    pty.paste_and_enter("REGISTRY_ONE");
    // Full-screen rendering may place ANSI cursor/style sequences between
    // words; the semantic reply is still visible as terminal text.
    pty.expect(&mut rendered, "REPLY");
    assert!(rendered.contains('0'), "{rendered}");
    pty.paste_and_enter("/detach");
    wait_success(&mut frontend, &pty, &mut rendered);
    assert_eq!(pty.termios(), before);
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 2);

    // Reinvoking the same source reuses its exact active runtime.
    let reused = start(&source_one);
    assert!(reused.status.success(), "{:?}", reused.stderr);
    assert!(String::from_utf8(reused.stdout)
        .unwrap()
        .contains("reused runtime"));
    assert_eq!(list_live_sessions(&home, &supercode_home).len(), 2);

    // The other session is independently driven through its tmux frontend.
    let send_tmux_line = |line: &str| {
        let target = format!("{second_tmux}:frontend");
        let pasted = format!("\u{1b}[200~{line}\u{1b}[201~");
        assert!(Command::new("tmux")
            .args(["send-keys", "-t", &target, "-l", &pasted])
            .status()
            .unwrap()
            .success());
        std::thread::sleep(Duration::from_millis(50));
        assert!(Command::new("tmux")
            .args(["send-keys", "-t", &target, "Enter"])
            .status()
            .unwrap()
            .success());
    };
    send_tmux_line("REGISTRY_TWO");
    let deadline = Instant::now() + Duration::from_secs(15);
    loop {
        let output = Command::new("tmux")
            .args([
                "capture-pane",
                "-p",
                "-S",
                "-100",
                "-t",
                &format!("{second_tmux}:frontend"),
            ])
            .output()
            .unwrap();
        if String::from_utf8_lossy(&output.stdout).contains("REGISTRY REPLY 1") {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "tmux frontend did not render reply"
        );
        std::thread::sleep(Duration::from_millis(20));
    }
    send_tmux_line("/detach");
    join_thread_bounded(provider_task, Duration::from_secs(3), "tmux provider");

    let first_path = PathBuf::from(first["persistence_location"].as_str().unwrap());
    let second_path = PathBuf::from(second["persistence_location"].as_str().unwrap());
    let close_runtime = |runtime_id: &str| {
        let mut close = Command::new(bin());
        close
            .env("HOME", &home)
            .env("SUPERCODE_HOME", &supercode_home)
            .args(["sessions", "close", runtime_id]);
        let output = output_bounded(&mut close, Duration::from_secs(10)).unwrap();
        assert!(output.status.success(), "{:?}", output.stderr);
    };
    for runtime_id in [&first_id, &second_id] {
        close_runtime(runtime_id);
    }
    assert!(list_live_sessions(&home, &supercode_home).is_empty());
    let gone_deadline = Instant::now() + Duration::from_secs(5);
    while [&first_tmux, &second_tmux].iter().any(|name| {
        Command::new("tmux")
            .args(["has-session", "-t", name])
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .status()
            .is_ok_and(|status| status.success())
    }) {
        assert!(Instant::now() < gone_deadline, "closed tmux owner survived");
        std::thread::sleep(Duration::from_millis(20));
    }

    // A tmux container with no registry receipt is presentation residue, not
    // a session. The next deterministic launch removes only that container,
    // starts a new owner, and leaves all persisted/source data untouched.
    assert!(Command::new("tmux")
        .args(["new-session", "-d", "-s", &first_tmux, "sleep", "30"])
        .status()
        .unwrap()
        .success());
    let reconciled = start(&source_one);
    assert!(reconciled.status.success(), "{:?}", reconciled.stderr);
    assert!(String::from_utf8(reconciled.stdout)
        .unwrap()
        .contains("started runtime"));
    let after_reconcile = list_live_sessions(&home, &supercode_home);
    assert_eq!(after_reconcile.len(), 1, "{after_reconcile:?}");
    close_runtime(after_reconcile[0]["runtime_id"].as_str().unwrap());
    assert!(list_live_sessions(&home, &supercode_home).is_empty());
    let hook_events = fs::read_to_string(&hooks_log).unwrap();
    assert_eq!(
        hook_events.matches("session_start").count(),
        3,
        "{hook_events}"
    );
    assert_eq!(
        hook_events.matches("session_end").count(),
        3,
        "{hook_events}"
    );
    assert_eq!(fs::read(&source_one).unwrap(), source_one_before);
    assert_eq!(fs::read(&source_two).unwrap(), source_two_before);
    let contents = |path: PathBuf| {
        fs::read_to_string(path)
            .unwrap()
            .lines()
            .filter(|line| !line.trim().is_empty())
            .map(|line| {
                serde_json::from_str::<supercode::ChatMessage>(line)
                    .unwrap()
                    .content
                    .unwrap_or_default()
            })
            .collect::<Vec<_>>()
    };
    let first_history = contents(first_path);
    let second_history = contents(second_path);
    assert!(
        first_history.windows(3).any(|messages| messages
            == [
                "imported tmux-source-one",
                "REGISTRY_ONE",
                "REGISTRY REPLY 0"
            ]),
        "{first_history:?}"
    );
    assert!(
        second_history.windows(3).any(|messages| messages
            == [
                "imported tmux-source-two",
                "REGISTRY_TWO",
                "REGISTRY REPLY 1"
            ]),
        "{second_history:?}"
    );
    fs::remove_dir_all(home).ok();
}