// m5_5_acceptance.rs --- Acceptance suite for M5.5 (daemon + attach).

//! End-to-end acceptance tests for T M5.5 (daemon mode +
//! local-socket transport).
//!
//! Each test spawns the real `pmacs` binary as a subprocess, drives
//! it via the published protocol crate, and tears it down on
//! completion. Tempdir-scoped sockets keep tests isolated under
//! parallel `cargo test`.
//!
//! Tests in this file (mapped to T M5.5 acceptance criteria):
//!
//! 1. [`daemon_starts_socket_and_lockfile_appear_with_correct_modes`]
//! 2. [`attach_send_key_receive_cell_response`]
//! 3. [`clean_detach_then_reattach`]
//! 4. [`ungraceful_disconnect_then_reattach`]
//! 5. [`second_daemon_same_socket_fails_clearly`]
//! 6. [`sigterm_daemon_sends_goodbye_and_cleans_up`]
//! 7. [`sigkill_daemon_leaves_stale_files_next_start_recovers`]
//! 8. [`version_mismatch_clean_disconnect`]

use std::fs;
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::thread;
use std::time::{Duration, Instant};

use nix::sys::signal::{Signal, kill};
use nix::unistd::Pid;
use tempfile::TempDir;

use pmacs::cell::CellSize;
#[cfg(feature = "crdt")]
use pmacs::cell::Color;
#[cfg(feature = "crdt")]
use pmacs::overlay_color::color_for_slot;
use pmacs::protocol::{
    AttachRequest, FrontendCapabilities, FrontendEvent, GoodbyeReason, Hello, InstanceMessage, Key,
    KeyEvent, Modifiers, PROTOCOL_VERSION,
};
use pmacs::transport::{read_message, write_message};

// ---------------------------------------------------------------------------
// Harness
// ---------------------------------------------------------------------------

/// A foreground daemon spawned in the test, with cleanup on Drop.
struct TestDaemon {
    /// Tempdir holding the socket and lockfile; auto-cleaned on Drop.
    _tempdir: TempDir,
    socket_path: PathBuf,
    process: Child,
}

impl TestDaemon {
    fn spawn() -> Self {
        Self::spawn_with_env(&[])
    }

    /// T M10.8 Day 4 — spawn with extra env-var overrides for
    /// instance-capability tests.
    fn spawn_with_env(env_vars: &[(&str, &str)]) -> Self {
        let tempdir = TempDir::new().expect("tempdir");
        // tempfile::TempDir creates 0755-mode directories; the daemon
        // requires a 0700-or-stricter parent for the socket. Tighten
        // the tempdir before spawning.
        fs::set_permissions(tempdir.path(), fs::Permissions::from_mode(0o700))
            .expect("chmod tempdir 0700");
        let socket_path = tempdir.path().join("pmacs.sock");
        let process = spawn_daemon_process_with_env(&socket_path, env_vars);
        wait_for_socket_or_exit(&socket_path, &process, Duration::from_secs(10))
            .expect("daemon socket appeared");
        Self {
            _tempdir: tempdir,
            socket_path,
            process,
        }
    }

    fn pid(&self) -> u32 {
        self.process.id()
    }

    fn connect(&self) -> UnixStream {
        UnixStream::connect(&self.socket_path).expect("connect")
    }

    fn is_alive(&mut self) -> bool {
        self.process.try_wait().ok().flatten().is_none()
    }

    fn lockfile_path(&self) -> PathBuf {
        let mut s = self.socket_path.as_os_str().to_os_string();
        s.push(".lock");
        PathBuf::from(s)
    }
}

impl Drop for TestDaemon {
    fn drop(&mut self) {
        let _ = self.process.kill();
        let _ = self.process.wait();
    }
}

fn spawn_daemon_process(socket_path: &Path) -> Child {
    spawn_daemon_process_with_env(socket_path, &[])
}

/// Spawn a daemon with extra environment-variable overrides.
///
/// T M10.8 Day 4 — used by tests that need to exercise non-default
/// instance capabilities (e.g., the M10.7 mismatch test, which
/// needs `PMACS_INSTANCE_MULTI_FRONTEND=0` so a frontend declaring
/// `multi_frontend: true` hits the capability-mismatch path).
/// Production daemons don't set these vars.
fn spawn_daemon_process_with_env(socket_path: &Path, env_vars: &[(&str, &str)]) -> Child {
    let isolated_home = socket_path.parent().expect("socket has parent");
    let mut cmd = Command::new(env!("CARGO_BIN_EXE_pmacs"));
    cmd.args(["--daemon", "--socket"])
        .arg(socket_path)
        .env("HOME", isolated_home)
        .env("XDG_CONFIG_HOME", isolated_home)
        .stdout(Stdio::null())
        .stderr(Stdio::null());
    for (key, value) in env_vars {
        cmd.env(key, value);
    }
    cmd.spawn().expect("spawn pmacs --daemon")
}

fn wait_for_socket_or_exit(
    socket: &Path,
    _process: &Child,
    deadline: Duration,
) -> Result<(), String> {
    // We probe by attempting to *connect*, not by file-exists. A stale
    // socket file from a previous (crashed) daemon can satisfy
    // `exists` without anyone actually listening — only a successful
    // connect proves the new daemon has run through `bind` and
    // `listen`. ECONNREFUSED on a stale socket retries until the new
    // daemon takes over.
    let start = Instant::now();
    while start.elapsed() < deadline {
        if let Ok(mut stream) = UnixStream::connect(socket) {
            // Read the Hello so the daemon's `send_message` succeeds
            // and doesn't log a "send Hello failed" warning to its
            // stderr (which our test runner inherits). After reading
            // we drop without sending AttachRequest; the daemon
            // observes the disconnect and falls back to accept.
            stream
                .set_read_timeout(Some(Duration::from_millis(500)))
                .ok();
            let _ = read_message::<Hello>(&mut stream);
            return Ok(());
        }
        thread::sleep(Duration::from_millis(20));
    }
    Err(format!(
        "daemon did not start listening on {} within {deadline:?}",
        socket.display()
    ))
}

fn build_default_caps() -> FrontendCapabilities {
    FrontendCapabilities {
        synchronized_output: true,
        unicode_smp: true,
        true_color: true,
        mouse: true,
        bracketed_paste: true,
        terminal_kind: Some("test".into()),
        multi_frontend: false,
        crdt_replica: false,
    }
}

/// Read the daemon's `Hello`, send our `AttachRequest`, return the Hello.
fn do_handshake(stream: &mut UnixStream) -> Hello {
    let hello: Hello = read_message(stream).expect("read Hello");
    assert_eq!(hello.protocol_version, PROTOCOL_VERSION);
    let req = AttachRequest {
        protocol_version: PROTOCOL_VERSION,
        frontend_capabilities: build_default_caps(),
        initial_size: CellSize::new(24, 80),
    };
    write_message(stream, &req).expect("write AttachRequest");
    hello
}

// ---------------------------------------------------------------------------
// Test 1
// ---------------------------------------------------------------------------

#[test]
fn daemon_starts_socket_and_lockfile_appear_with_correct_modes() {
    let daemon = TestDaemon::spawn();

    // Socket: owner-only. The kernel applies umask 0077 to a base
    // 0777 for socket files (the lockfile is created via O_CREAT
    // with explicit mode and lands at 0600). The "x" bit on the
    // socket has no semantic meaning, so we assert the security
    // property — no group/other bits — rather than an exact 0o600.
    let socket_meta = fs::metadata(&daemon.socket_path).expect("stat socket");
    let socket_mode = socket_meta.permissions().mode() & 0o7777;
    assert_eq!(
        socket_mode & 0o077,
        0,
        "socket mode {socket_mode:#o} should be owner-only"
    );

    // Lockfile: mode 0600.
    let lockfile_path = daemon.lockfile_path();
    let lock_meta = fs::metadata(&lockfile_path).expect("stat lockfile");
    let lock_mode = lock_meta.permissions().mode() & 0o7777;
    assert_eq!(
        lock_mode, 0o600,
        "lockfile mode should be 0600, got {lock_mode:#o}"
    );

    // Parent dir: at most 0700 (no group/other bits).
    let parent_meta = fs::metadata(daemon.socket_path.parent().unwrap()).expect("stat parent");
    let parent_mode = parent_meta.permissions().mode() & 0o7777;
    assert_eq!(
        parent_mode & 0o077,
        0,
        "parent dir mode {parent_mode:#o} should be owner-only"
    );
}

// ---------------------------------------------------------------------------
// Test 2
// ---------------------------------------------------------------------------

#[test]
fn attach_send_key_receive_cell_response() {
    let daemon = TestDaemon::spawn();
    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();

    let hello = do_handshake(&mut stream);

    // After AttachRequest the daemon emits the initial full-grid sync.
    let initial: InstanceMessage = read_message(&mut stream).expect("initial frame");
    match initial {
        InstanceMessage::CellDelta {
            full_grid: true, ..
        } => {}
        other => panic!("expected initial full-grid CellDelta, got {other:?}"),
    }

    // Send a key event.
    let key_event = FrontendEvent::Key(KeyEvent {
        frontend_id: hello.assigned_frontend_id,
        key: Key::Char('a'),
        mods: Modifiers::NONE,
        timestamp_ns: 0,
    });
    write_message(&mut stream, &key_event).expect("send key");

    // The daemon should produce at least one render message in response
    // (cell delta or cursor update). Read up to ~2s for any message.
    let deadline = Instant::now() + Duration::from_secs(2);
    let mut got_response = false;
    while Instant::now() < deadline {
        stream
            .set_read_timeout(Some(Duration::from_millis(200)))
            .unwrap();
        if let Ok(msg) = read_message::<InstanceMessage>(&mut stream) {
            if matches!(
                msg,
                InstanceMessage::CellDelta { .. } | InstanceMessage::Cursor(_)
            ) {
                got_response = true;
                break;
            }
        }
    }
    assert!(got_response, "expected render response after key");
}

// ---------------------------------------------------------------------------
// Test 3
// ---------------------------------------------------------------------------

#[test]
fn clean_detach_then_reattach() {
    let mut daemon = TestDaemon::spawn();
    let pid_before = daemon.pid();

    // Attach, read initial frame, send Detach, drop.
    {
        let mut stream = daemon.connect();
        stream
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let hello = do_handshake(&mut stream);
        let _: InstanceMessage = read_message(&mut stream).expect("initial frame");
        write_message(
            &mut stream,
            &FrontendEvent::Detach(hello.assigned_frontend_id),
        )
        .expect("send Detach");
        drop(stream);
    }

    // Give the daemon time to clear its attached slot.
    thread::sleep(Duration::from_millis(300));

    // Reattach should succeed.
    {
        let mut stream = daemon.connect();
        stream
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let _hello = do_handshake(&mut stream);
        let _: InstanceMessage = read_message(&mut stream).expect("initial frame");
    }

    assert_eq!(daemon.pid(), pid_before);
    assert!(daemon.is_alive(), "daemon should still be running");
}

// ---------------------------------------------------------------------------
// Test 4
// ---------------------------------------------------------------------------

#[test]
fn ungraceful_disconnect_then_reattach() {
    let mut daemon = TestDaemon::spawn();
    let pid_before = daemon.pid();

    // Attach, do handshake, drop without Detach (simulates SIGKILL'd
    // frontend).
    {
        let mut stream = daemon.connect();
        stream
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let _ = do_handshake(&mut stream);
        // Note: we don't even read the initial frame — close immediately.
        drop(stream);
    }

    // Daemon needs longer to detect ungraceful close because the read
    // path has no deterministic wakeup; the per-attach loop polls the
    // channel with a frame-target timeout (~16 ms by default), so 500
    // ms is safely above any realistic detection latency.
    thread::sleep(Duration::from_millis(500));

    // Reattach should succeed.
    {
        let mut stream = daemon.connect();
        stream
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let _hello = do_handshake(&mut stream);
        let _: InstanceMessage = read_message(&mut stream).expect("initial frame");
    }

    assert_eq!(daemon.pid(), pid_before);
    assert!(daemon.is_alive(), "daemon should still be running");
}

// ---------------------------------------------------------------------------
// Test 5
// ---------------------------------------------------------------------------

#[test]
fn second_daemon_same_socket_fails_clearly() {
    let mut daemon_a = TestDaemon::spawn();

    // Spawn second daemon at the same socket; capture its stderr.
    let isolated_home = daemon_a.socket_path.parent().unwrap();
    let output = Command::new(env!("CARGO_BIN_EXE_pmacs"))
        .args(["--daemon", "--socket"])
        .arg(&daemon_a.socket_path)
        .env("HOME", isolated_home)
        .env("XDG_CONFIG_HOME", isolated_home)
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .output()
        .expect("spawn second daemon");

    assert!(
        !output.status.success(),
        "second daemon should exit non-zero"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("daemon already running"),
        "expected error message in stderr, got: {stderr}"
    );

    // First daemon unaffected: still alive, still accepts handshake.
    assert!(daemon_a.is_alive(), "daemon A should still be running");
    let mut stream = daemon_a.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let _hello = do_handshake(&mut stream);
}

// ---------------------------------------------------------------------------
// Test 6
// ---------------------------------------------------------------------------

#[test]
fn sigterm_daemon_sends_goodbye_and_cleans_up() {
    let mut daemon = TestDaemon::spawn();
    let socket_path = daemon.socket_path.clone();
    let lockfile_path = daemon.lockfile_path();
    let pid = daemon.pid();

    // Attach.
    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let _hello = do_handshake(&mut stream);
    let _: InstanceMessage = read_message(&mut stream).expect("initial frame");

    // Send SIGTERM to daemon.
    kill(Pid::from_raw(i32::try_from(pid).unwrap()), Signal::SIGTERM).expect("kill SIGTERM");

    // Read until we see Goodbye(ShuttingDown) or hit EOF.
    stream
        .set_read_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    let mut got_goodbye = false;
    loop {
        match read_message::<InstanceMessage>(&mut stream) {
            Ok(InstanceMessage::Goodbye(GoodbyeReason::ShuttingDown)) => {
                got_goodbye = true;
                break;
            }
            // Intermediate render frames are fine.
            Ok(_) => {}
            // EOF / error — daemon closed.
            Err(_) => break,
        }
    }
    assert!(got_goodbye, "expected Goodbye(ShuttingDown) before EOF");

    // Daemon should exit cleanly.
    let status = daemon.process.wait().expect("wait");
    assert!(
        status.success(),
        "daemon should exit 0 after SIGTERM, got {status}"
    );

    // Both files cleaned up.
    assert!(
        !socket_path.exists(),
        "socket {socket_path:?} should be unlinked"
    );
    assert!(
        !lockfile_path.exists(),
        "lockfile {lockfile_path:?} should be unlinked"
    );
}

// ---------------------------------------------------------------------------
// Test 7
// ---------------------------------------------------------------------------

#[test]
fn sigkill_daemon_leaves_stale_files_next_start_recovers() {
    let tempdir = TempDir::new().expect("tempdir");
    fs::set_permissions(tempdir.path(), fs::Permissions::from_mode(0o700))
        .expect("chmod tempdir 0700");
    let socket_path = tempdir.path().join("pmacs.sock");

    // First daemon.
    let mut daemon1 = spawn_daemon_process(&socket_path);
    wait_for_socket_or_exit(&socket_path, &daemon1, Duration::from_secs(10))
        .expect("daemon 1 socket appeared");

    let mut lockfile_path = socket_path.as_os_str().to_os_string();
    lockfile_path.push(".lock");
    let lockfile_path = PathBuf::from(lockfile_path);

    assert!(socket_path.exists(), "socket exists pre-SIGKILL");
    assert!(lockfile_path.exists(), "lockfile exists pre-SIGKILL");

    // SIGKILL the daemon.
    kill(
        Pid::from_raw(i32::try_from(daemon1.id()).unwrap()),
        Signal::SIGKILL,
    )
    .expect("SIGKILL");
    let _ = daemon1.wait();

    // Stale files persist (kernel only released the flock + socket fd).
    assert!(
        socket_path.exists(),
        "socket should persist on disk after SIGKILL"
    );
    assert!(
        lockfile_path.exists(),
        "lockfile should persist on disk after SIGKILL"
    );

    // Second daemon must successfully recover.
    let mut daemon2 = spawn_daemon_process(&socket_path);
    wait_for_socket_or_exit(&socket_path, &daemon2, Duration::from_secs(10))
        .expect("daemon 2 socket appeared");

    // Verify socket is fresh and owner-only (kernel applies umask
    // 0077 to the implicit 0777 base for socket files → 0700; the
    // "x" bit is meaningless on a socket; the security property is
    // "no group/other access").
    let mode = fs::metadata(&socket_path).unwrap().permissions().mode() & 0o7777;
    assert_eq!(
        mode & 0o077,
        0,
        "socket mode {mode:#o} should be owner-only"
    );

    // Lockfile pid was rewritten.
    let pid_str = fs::read_to_string(&lockfile_path).unwrap();
    let parsed: u32 = pid_str.trim().parse().expect("pid in lockfile");
    assert_eq!(parsed, daemon2.id(), "lockfile pid should match daemon 2");

    // Cleanup.
    let _ = daemon2.kill();
    let _ = daemon2.wait();
}

// ---------------------------------------------------------------------------
// Test 8
// ---------------------------------------------------------------------------

#[test]
fn version_mismatch_clean_disconnect() {
    let mut daemon = TestDaemon::spawn();

    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();

    // Read Hello.
    let hello: Hello = read_message(&mut stream).expect("Hello");
    assert_eq!(hello.protocol_version, PROTOCOL_VERSION);

    // Send AttachRequest with wrong protocol version.
    let req = AttachRequest {
        protocol_version: 999,
        frontend_capabilities: build_default_caps(),
        initial_size: CellSize::new(24, 80),
    };
    write_message(&mut stream, &req).expect("write");

    // Expect Goodbye(VersionMismatch).
    match read_message::<InstanceMessage>(&mut stream) {
        Ok(InstanceMessage::Goodbye(GoodbyeReason::VersionMismatch { server, client })) => {
            assert_eq!(server, PROTOCOL_VERSION);
            assert_eq!(client, 999);
        }
        other => panic!("expected VersionMismatch Goodbye, got {other:?}"),
    }

    drop(stream);

    // Daemon still alive and serving.
    assert!(daemon.is_alive());
    let mut stream2 = daemon.connect();
    stream2
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let _hello = do_handshake(&mut stream2);
}

// ---------------------------------------------------------------------------
// T M10.7 — capability negotiation, daemon end-to-end.
//
// Pure-function tests for `negotiate_capabilities` live in
// `src/protocol.rs` and exercise the negotiation logic exhaustively.
// This end-to-end test exercises the wire path: a v2 frontend
// declaring `multi_frontend: true` receives `Goodbye(
// CapabilityMismatch)` from a v2 daemon whose
// `InstanceCapabilities::default()` has `multi_frontend: false`
// (M10.8 flips that bit; M10.7 ships the mismatch path).
// ---------------------------------------------------------------------------

#[test]
fn m10_7_capability_mismatch_v2_frontend_wants_multi_frontend() {
    // T M10.8 Day 4 rewrite: M10.8 flipped `InstanceCapabilities::
    // default()` to advertise `multi_frontend: true` and `crdt_replica:
    // true`. To preserve daemon-end-to-end coverage of the M10.7
    // mismatch path, this test spawns the daemon with
    // `PMACS_INSTANCE_MULTI_FRONTEND=0` so the instance advertises
    // `multi_frontend: false` and a frontend declaring `true`
    // hits the mismatch.
    //
    // Approach (i) from the M10.8 framing-pass review: explicit
    // instance-caps override preserves daemon-end-to-end coverage.
    // Pure-function negotiation tests in `protocol.rs` still cover
    // the negotiation logic itself; this test specifically
    // exercises the daemon's wiring of negotiate_capabilities into
    // handle_connection.
    let mut daemon = TestDaemon::spawn_with_env(&[("PMACS_INSTANCE_MULTI_FRONTEND", "0")]);
    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();

    let _hello: Hello = read_message(&mut stream).expect("read Hello");

    let caps_multi = FrontendCapabilities {
        multi_frontend: true,
        ..build_default_caps()
    };
    let req = AttachRequest {
        protocol_version: PROTOCOL_VERSION,
        frontend_capabilities: caps_multi,
        initial_size: CellSize::new(24, 80),
    };
    write_message(&mut stream, &req).expect("write AttachRequest");

    match read_message::<InstanceMessage>(&mut stream) {
        Ok(InstanceMessage::Goodbye(GoodbyeReason::CapabilityMismatch { missing })) => {
            assert_eq!(
                missing,
                vec!["multi_frontend".to_string()],
                "M10.7 criterion 4: error names the requested capability"
            );
        }
        other => panic!("expected Goodbye(CapabilityMismatch), got {other:?}"),
    }

    // Daemon stays up; another connection without multi_frontend
    // succeeds even with the env-var override (the env var only
    // affects the instance's advertised caps, not what frontends
    // request — a frontend not asking for multi_frontend doesn't
    // hit the mismatch).
    drop(stream);
    assert!(daemon.is_alive());
    let mut stream2 = daemon.connect();
    stream2
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let _hello2 = do_handshake(&mut stream2);
}

#[test]
fn m10_7_no_negotiation_for_v1_frontend() {
    // T M10.7 criterion 1 — a v0.1 frontend (no `multi_frontend`
    // declared in the wire format) attaches successfully. The
    // negotiation function sees frontend.multi_frontend = false
    // (from #[serde(default)]) and produces a clean Ok(...) result;
    // no mismatch is generated.
    let daemon = TestDaemon::spawn();
    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();

    let _hello = do_handshake(&mut stream);
    // do_handshake's default caps have multi_frontend: false; the
    // attach should succeed and produce the initial CellDelta. If
    // the daemon had emitted Goodbye(CapabilityMismatch) instead,
    // read_message below would decode that variant — but the
    // standard handshake path produces the CellDelta.
    let initial: InstanceMessage = read_message(&mut stream).expect("initial frame");
    match initial {
        InstanceMessage::CellDelta {
            full_grid: true, ..
        } => {}
        other => panic!("expected initial full-grid CellDelta, got {other:?}"),
    }
}

// ---------------------------------------------------------------------------
// T M10.8 Day 4 — multi-attach end-to-end acceptance + Q5 admission matrix.
// ---------------------------------------------------------------------------

/// Caps for a v1.0 multi-frontend frontend (declares both bits).
#[cfg(feature = "crdt")]
fn multi_frontend_caps() -> FrontendCapabilities {
    FrontendCapabilities {
        multi_frontend: true,
        crdt_replica: true,
        ..build_default_caps()
    }
}

/// Connect and run the `AttachRequest` with multi-frontend caps.
/// Reads but does not assert on the initial `CellDelta` — caller
/// handles that. Returns the Hello + stream.
#[cfg(feature = "crdt")]
fn attach_multi(daemon: &TestDaemon) -> (Hello, UnixStream) {
    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let hello: Hello = read_message(&mut stream).expect("read Hello");
    let req = AttachRequest {
        protocol_version: PROTOCOL_VERSION,
        frontend_capabilities: multi_frontend_caps(),
        initial_size: CellSize::new(24, 80),
    };
    write_message(&mut stream, &req).expect("write AttachRequest");
    (hello, stream)
}

/// M10.8 acceptance criterion 1 + 2 + 3 — happy-path multi-attach.
///
/// Two v1.0 frontends both negotiate `multi_frontend: true`,
/// attach simultaneously, and each receives at least one
/// presence broadcast from the other when the other moves its
/// cursor.
// Multi-frontend tests below exercise the M10.8/M10.9 multi-frontend
// + presence + CRDT path. Post-audit Finding 3 fix made
// `InstanceCapabilities::default()` advertise `multi_frontend: false`
// in non-CRDT builds (because multi-frontend's payoff is the CRDT
// pipeline). These tests are inherently CRDT-feature-only.
#[cfg(feature = "crdt")]
#[test]
fn m10_8_two_frontends_attached_simultaneously_receive_presence_broadcasts() {
    let daemon = TestDaemon::spawn();

    let (hello_a, mut stream_a) = attach_multi(&daemon);
    let (hello_b, mut stream_b) = attach_multi(&daemon);
    assert_ne!(hello_a.assigned_frontend_id, hello_b.assigned_frontend_id);

    // Read initial frames synchronously so kernel buffers stay
    // drained; drain_pending's 50ms-timeout reads back up under
    // the dispatcher's per-tick writes.
    let _initial_a: InstanceMessage = read_message(&mut stream_a).expect("A initial frame");
    let _initial_b: InstanceMessage = read_message(&mut stream_b).expect("B initial frame");

    // Send a key event from A. The daemon dispatches it to A's
    // active window; cursor moves; per-tick presence sweep
    // produces a PresenceUpdate broadcast to B.
    let key = FrontendEvent::Key(pmacs::protocol::KeyEvent {
        frontend_id: hello_a.assigned_frontend_id,
        key: pmacs::protocol::Key::Char('x'),
        mods: pmacs::protocol::Modifiers::NONE,
        timestamp_ns: 0,
    });
    write_message(&mut stream_a, &key).expect("send key from A");

    // Both A and B will receive some messages. We're specifically
    // checking that B receives a PresenceUpdate sourced from
    // hello_a.assigned_frontend_id.
    let deadline = Instant::now() + Duration::from_secs(2);
    stream_b
        .set_read_timeout(Some(Duration::from_millis(200)))
        .unwrap();
    let mut saw_presence_from_a = false;
    while Instant::now() < deadline {
        if let Ok(InstanceMessage::PresenceUpdate { frontend_id, .. }) =
            read_message::<InstanceMessage>(&mut stream_b)
            && frontend_id == hello_a.assigned_frontend_id
        {
            saw_presence_from_a = true;
            break;
        }
        // Other messages (CellDelta, Cursor) and read timeouts both
        // just continue the deadline loop.
    }
    assert!(
        saw_presence_from_a,
        "M10.8 criterion 1: frontend B should receive a PresenceUpdate from A's cursor move"
    );
}

/// M10.8 acceptance criterion 2 — disconnect is local.
///
/// Attach A and B; disconnect A; verify B continues to operate.
#[cfg(feature = "crdt")]
#[test]
fn m10_8_disconnect_of_one_frontend_does_not_affect_the_other() {
    let mut daemon = TestDaemon::spawn();

    let (_hello_a, stream_a) = attach_multi(&daemon);
    let (hello_b, mut stream_b) = attach_multi(&daemon);
    let _initial_b: InstanceMessage = read_message(&mut stream_b).expect("B initial frame");

    // Disconnect A by dropping its stream.
    drop(stream_a);

    // Give the dispatcher a moment to process A's detach.
    thread::sleep(Duration::from_millis(100));

    // B can still send events and the daemon responds.
    let key = FrontendEvent::Key(pmacs::protocol::KeyEvent {
        frontend_id: hello_b.assigned_frontend_id,
        key: pmacs::protocol::Key::Char('y'),
        mods: pmacs::protocol::Modifiers::NONE,
        timestamp_ns: 0,
    });
    write_message(&mut stream_b, &key).expect("B can still send keys");

    stream_b
        .set_read_timeout(Some(Duration::from_secs(2)))
        .unwrap();
    let response: InstanceMessage =
        read_message(&mut stream_b).expect("B receives response after A disconnected");
    // Any response counts — CellDelta from B's edit, Cursor, or a
    // presence broadcast from before the disconnect. The
    // assertion is that B's connection is still live.
    let _ = response;
    assert!(daemon.is_alive());
}

/// Q5 row 1: v0.1 frontend attached; another v0.1 attaches; rejected.
#[test]
fn m10_8_q5_row1_v01_with_v01_attempt_rejected() {
    let daemon = TestDaemon::spawn();

    // First v0.1 frontend (no multi_frontend, no crdt_replica): attaches.
    let mut stream_a = daemon.connect();
    stream_a
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let _hello_a = do_handshake(&mut stream_a);
    // Read initial frame so the daemon's per-tick render isn't
    // backed up on stream_a (which might trigger spurious write
    // failures elsewhere).
    let initial: InstanceMessage = read_message(&mut stream_a).expect("initial frame on A");
    let _ = initial;

    // Second v0.1 frontend: rejected with AlreadyAttached.
    let mut stream_b = daemon.connect();
    stream_b
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let _hello_b: Hello = read_message(&mut stream_b).expect("read Hello");
    let req_b = AttachRequest {
        protocol_version: PROTOCOL_VERSION,
        frontend_capabilities: build_default_caps(),
        initial_size: CellSize::new(24, 80),
    };
    write_message(&mut stream_b, &req_b).expect("write AttachRequest");

    match read_message::<InstanceMessage>(&mut stream_b) {
        Ok(InstanceMessage::Goodbye(GoodbyeReason::AlreadyAttached)) => {}
        other => panic!("Q5 row 1: expected AlreadyAttached, got {other:?}"),
    }
}

/// Q5 row 2: v0.1 attached; v2 multi attaches; both coexist
/// (heterogeneous case; the v0.1 sees its session normally, the
/// v2 sees the v0.1 as a "ghost editor").
#[cfg(feature = "crdt")]
#[test]
fn m10_8_q5_row2_v01_with_v2_multi_coexist() {
    let daemon = TestDaemon::spawn();

    // First: v0.1 frontend attaches normally.
    let mut stream_v1 = daemon.connect();
    stream_v1
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let _hello_v1 = do_handshake(&mut stream_v1);
    // Read initial frame synchronously so the kernel buffer
    // doesn't back up; using drain_pending here causes the
    // dispatcher's per-tick writes to accumulate while the test
    // is mid-50ms-timeout, which can leave streams in a flaky
    // state by the time the second attach runs.
    let _initial_v1: InstanceMessage = read_message(&mut stream_v1).expect("v0.1 initial frame");

    // Second: v2 multi-capable frontend attaches alongside.
    let (_hello_v2, mut stream_v2) = attach_multi(&daemon);
    // If the multi attach succeeded, the daemon sends initial
    // CellDelta etc. — read at least one frame to confirm
    // acceptance.
    let initial: InstanceMessage = read_message(&mut stream_v2).expect("v2 initial frame");
    match initial {
        InstanceMessage::CellDelta { .. } => {}
        InstanceMessage::Goodbye(reason) => {
            panic!("Q5 row 2: v2 multi should attach alongside v0.1, got Goodbye: {reason:?}")
        }
        _other => {} // Cursor / etc. also acceptable as "attached"
    }
}

/// Q5 row 3: v2 multi attached; v0.1 attempts to attach; accepted
/// (Q5 logic: a non-multi attach is rejected iff
/// `count_non_multi_sessions > 0`; with only multi attached the
/// count is 0, so v0.1 takes the non-multi slot).
#[cfg(feature = "crdt")]
#[test]
fn m10_8_q5_row3_v2_multi_then_v01_accepted() {
    let daemon = TestDaemon::spawn();

    let (_hello_v2, mut stream_v2) = attach_multi(&daemon);
    // Read the v2 initial frame synchronously (same rationale as
    // Q5 row 2 — avoid drain_pending's 50ms-timeout buffer race).
    let _initial_v2: InstanceMessage = read_message(&mut stream_v2).expect("v2 initial frame");

    // v0.1 frontend attaches: should succeed because the multi
    // session doesn't occupy the non-multi slot.
    let mut stream_v1 = daemon.connect();
    stream_v1
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let _hello_v1: Hello = read_message(&mut stream_v1).expect("read Hello");
    let req_v1 = AttachRequest {
        protocol_version: PROTOCOL_VERSION,
        frontend_capabilities: build_default_caps(),
        initial_size: CellSize::new(24, 80),
    };
    write_message(&mut stream_v1, &req_v1).expect("write AttachRequest");

    let initial: InstanceMessage = read_message(&mut stream_v1).expect("v0.1 initial frame");
    match initial {
        InstanceMessage::CellDelta { .. } => {}
        InstanceMessage::Goodbye(reason) => {
            panic!("Q5 row 3: v0.1 should attach alongside v2 multi, got Goodbye: {reason:?}")
        }
        _other => {}
    }
}

/// Q5 row 4: two v2 multi sessions, both attached.
#[cfg(feature = "crdt")]
#[test]
fn m10_8_q5_row4_two_v2_multi_sessions_attached() {
    let daemon = TestDaemon::spawn();
    let (hello_a, mut stream_a) = attach_multi(&daemon);
    let (hello_b, mut stream_b) = attach_multi(&daemon);
    assert_ne!(hello_a.assigned_frontend_id, hello_b.assigned_frontend_id);

    // Both receive initial frames; no rejection.
    let _frame_a: InstanceMessage = read_message(&mut stream_a).expect("A initial");
    let _frame_b: InstanceMessage = read_message(&mut stream_b).expect("B initial");
}

// ---------------------------------------------------------------------------
// T M10.9 — overlay color rendering, daemon end-to-end.
// ---------------------------------------------------------------------------

/// M10.9 acceptance criterion 1: two frontends in the same buffer
/// see each other's cursors.
///
/// Spawn daemon, attach two multi-capable frontends. They share
/// LOCAL's buffer (M10.9 attach behavior). A inserts a few
/// characters (moving A's cursor); B's subsequent `CellDelta`
/// should contain an overlay cell with A's assigned color.
#[cfg(feature = "crdt")]
#[test]
fn m10_9_other_frontend_cursor_appears_in_recipient_cell_delta_with_color() {
    let daemon = TestDaemon::spawn();
    let (hello_a, mut stream_a) = attach_multi(&daemon);
    let (_hello_b, mut stream_b) = attach_multi(&daemon);

    // Read initial CellDelta + Cursor on each. M10.9: B's initial
    // frame includes A's cursor at byte 0 (where A's window starts).
    let _initial_a: InstanceMessage = read_message(&mut stream_a).expect("A initial");
    let _initial_b_1: InstanceMessage = read_message(&mut stream_b).expect("B initial CellDelta");
    let _initial_b_2: InstanceMessage = read_message(&mut stream_b).expect("B initial Cursor");

    // A presses a character. A's cursor advances; B should
    // receive a CellDelta that includes an overlay cell with A's
    // color.
    let key = FrontendEvent::Key(KeyEvent {
        frontend_id: hello_a.assigned_frontend_id,
        key: Key::Char('z'),
        mods: Modifiers::NONE,
        timestamp_ns: 0,
    });
    write_message(&mut stream_a, &key).expect("send key from A");

    // Compute A's expected color. With both A and B from same uid
    // (test process), they share a color slot — so this test
    // verifies the *presence* of an overlay-style cell, not a
    // specific color. We assert that B's CellDelta contains at
    // least one cell whose fg is in the M10.9 palette.
    let deadline = Instant::now() + Duration::from_secs(2);
    stream_b
        .set_read_timeout(Some(Duration::from_millis(200)))
        .unwrap();
    let mut saw_palette_cell = false;
    while Instant::now() < deadline {
        if let Ok(InstanceMessage::CellDelta { spans, .. }) =
            read_message::<InstanceMessage>(&mut stream_b)
        {
            for span in &spans {
                for cell in &span.cells {
                    // The palette uses Color::Rgb(...). Any cell whose
                    // fg is a palette entry is an overlay cell.
                    if let Color::Rgb(_, _, _) = cell.style.fg {
                        if is_palette_color(cell.style.fg) {
                            saw_palette_cell = true;
                            break;
                        }
                    }
                }
                if saw_palette_cell {
                    break;
                }
            }
        }
        if saw_palette_cell {
            break;
        }
    }
    assert!(
        saw_palette_cell,
        "M10.9 criterion 1: B's CellDelta should contain an overlay cell with a palette color after A moved its cursor"
    );
}

#[cfg(feature = "crdt")]
fn is_palette_color(c: Color) -> bool {
    for slot in 0..pmacs::overlay_color::PALETTE_LEN as u8 {
        if color_for_slot(slot) == c {
            return true;
        }
    }
    false
}

/// M10.9 acceptance criterion 2 (partial): color stability for
/// same-uid reconnect.
///
/// Attach A → detach A → reattach A'. A and A' connect from the
/// same uid (the test process). The daemon's `color_registry` maps
/// uid → slot. The slot should be the same for both attaches.
///
/// This test verifies the color-stability property at the daemon
/// level by attaching B as observer, then attaching A (slot
/// recorded), detaching A, reattaching A' (slot reused from
/// uid lookup), and checking B sees consistent overlay color.
#[cfg(feature = "crdt")]
#[test]
fn m10_9_color_stable_across_reconnect_for_same_uid() {
    let daemon = TestDaemon::spawn();
    // B is the observer.
    let (_hello_b, mut stream_b) = attach_multi(&daemon);
    let _initial_b_cd: InstanceMessage = read_message(&mut stream_b).expect("B initial cd");
    let _initial_b_cu: InstanceMessage = read_message(&mut stream_b).expect("B initial cursor");

    // A attaches the first time.
    let (hello_a1, mut stream_a1) = attach_multi(&daemon);
    let _a1_init: InstanceMessage = read_message(&mut stream_a1).expect("A1 init");
    // A1 moves cursor so an overlay is generated for B.
    let key1 = FrontendEvent::Key(KeyEvent {
        frontend_id: hello_a1.assigned_frontend_id,
        key: Key::Char('x'),
        mods: Modifiers::NONE,
        timestamp_ns: 0,
    });
    write_message(&mut stream_a1, &key1).expect("send key from A1");

    // Capture A1's overlay color from B.
    let color_a1 = wait_for_palette_color_in_b(&mut stream_b, Duration::from_secs(2))
        .expect("should observe A1's overlay color");

    // Detach A1.
    drop(stream_a1);
    thread::sleep(Duration::from_millis(100));

    // A2 reattaches (same test process, same uid).
    let (hello_a2, mut stream_a2) = attach_multi(&daemon);
    let _a2_init: InstanceMessage = read_message(&mut stream_a2).expect("A2 init");
    let key2 = FrontendEvent::Key(KeyEvent {
        frontend_id: hello_a2.assigned_frontend_id,
        key: Key::Char('y'),
        mods: Modifiers::NONE,
        timestamp_ns: 0,
    });
    write_message(&mut stream_a2, &key2).expect("send key from A2");

    let color_a2 = wait_for_palette_color_in_b(&mut stream_b, Duration::from_secs(2))
        .expect("should observe A2's overlay color");

    assert_eq!(
        color_a1, color_a2,
        "M10.9 criterion 2: same uid across reconnect → same color slot"
    );
}

/// Helper for color-stability test: read `CellDelta` messages from
/// `stream` for up to `timeout` and return the first palette
/// color found in any overlay cell.
#[cfg(feature = "crdt")]
fn wait_for_palette_color_in_b(stream: &mut UnixStream, timeout: Duration) -> Option<Color> {
    let deadline = Instant::now() + timeout;
    stream
        .set_read_timeout(Some(Duration::from_millis(200)))
        .ok();
    while Instant::now() < deadline {
        if let Ok(InstanceMessage::CellDelta { spans, .. }) =
            read_message::<InstanceMessage>(stream)
        {
            for span in &spans {
                for cell in &span.cells {
                    if let Color::Rgb(_, _, _) = cell.style.fg {
                        if is_palette_color(cell.style.fg) {
                            return Some(cell.style.fg);
                        }
                    }
                }
            }
        }
    }
    None
}

// ---------------------------------------------------------------------------
// T M10.10 Day 2 — daemon sends BufferSnapshot on SessionEstablished
// for crdt_replica-negotiated frontends; not for non-replica frontends.
// ---------------------------------------------------------------------------

/// M10.10 Day 2 acceptance: a frontend that negotiated
/// `crdt_replica: true` receives an `InstanceMessage::BufferSnapshot`
/// for the `*scratch*` buffer immediately after `SessionEstablished`,
/// before any `CellDelta`. The snapshot bytes round-trip into a fresh
/// `CrdtState` and yield the expected materialized content.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_replica_frontend_receives_buffer_snapshot_before_cell_delta() {
    let daemon = TestDaemon::spawn();
    let (_hello, mut stream) = attach_multi(&daemon);

    // The first InstanceMessage after handshake must be a
    // BufferSnapshot. The dispatcher sends snapshots before the first
    // per-tick render fires.
    let first: InstanceMessage = read_message(&mut stream).expect("first frame");
    let (buffer_id, snapshot_bytes) = match first {
        InstanceMessage::BufferSnapshot {
            buffer_id,
            crdt_snapshot,
        } => (buffer_id, crdt_snapshot),
        other => {
            panic!("M10.10: first frame for replica frontend must be BufferSnapshot, got {other:?}")
        }
    };
    assert!(
        !snapshot_bytes.is_empty(),
        "snapshot bytes must be non-empty (loro encodes empty state as a non-empty payload)"
    );

    // The snapshot must decode into a fresh CrdtState. peer_id 0xBEEF
    // here is arbitrary — bootstrap on the frontend uses
    // peer_id_from_frontend(my_id), but the round-trip test doesn't
    // care which peer reconstructs.
    let replica = pmacs::crdt::CrdtState::new(0xBEEF).expect("fresh CrdtState");
    replica
        .import_snapshot(&snapshot_bytes)
        .expect("import the daemon's snapshot");
    // The daemon's *scratch* buffer starts empty.
    assert_eq!(replica.materialize_string(), "");
    let _ = buffer_id; // consumed for the panic-message in the match
}

/// M10.10 Day 3 Finding 2 acceptance: a replica frontend receives an
/// authoritative `InstanceMessage::CursorByte` paired with the regular
/// `Cursor` grid update. The byte position is the daemon's
/// active-window cursor at the moment of the render frame.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_replica_frontend_receives_cursor_byte_paired_with_cursor() {
    let daemon = TestDaemon::spawn();
    let (_hello, mut stream) = attach_multi(&daemon);

    // Drain initial frames until both Cursor and CursorByte have
    // been seen. The daemon emits per-tick: CellDelta, Cursor,
    // CursorByte. BufferSnapshot fires before all of those at
    // session establishment. We want to confirm that for the same
    // render iteration, Cursor and CursorByte both arrive.
    stream
        .set_read_timeout(Some(Duration::from_secs(2)))
        .unwrap();
    let mut saw_cursor = false;
    let mut saw_cursor_byte = false;
    for _ in 0..16 {
        match read_message::<InstanceMessage>(&mut stream) {
            Ok(InstanceMessage::Cursor(_)) => saw_cursor = true,
            Ok(InstanceMessage::CursorByte { .. }) => saw_cursor_byte = true,
            Ok(_) => continue,
            Err(_) => break,
        }
        if saw_cursor && saw_cursor_byte {
            break;
        }
    }
    assert!(saw_cursor, "replica frontend should receive Cursor");
    assert!(
        saw_cursor_byte,
        "replica frontend should receive CursorByte alongside Cursor (Finding 2)"
    );
}

/// M10.10 Day 3 Finding 3 acceptance: end-to-end `CrdtOp` broadcast.
/// Frontend A (replica) sends `FrontendEvent::CrdtOp` to the daemon;
/// daemon applies it to the buffer's CRDT + rope, then fans out
/// `InstanceMessage::CrdtOp` to other replica frontends. Frontend B
/// (also replica) receives the broadcast tagged with A's
/// `frontend_id` so B's mirror can route through the echo-dedup
/// filter (apply, not skip).
#[cfg(feature = "crdt")]
#[test]
fn m10_10_crdt_op_from_a_reaches_b_via_daemon_broadcast() {
    let daemon = TestDaemon::spawn();

    // Attach A and B as replica frontends.
    let (hello_a, mut stream_a) = attach_multi(&daemon);
    let (hello_b, mut stream_b) = attach_multi(&daemon);
    assert_ne!(hello_a.assigned_frontend_id, hello_b.assigned_frontend_id);

    // Drain A's first frame (BufferSnapshot) and bootstrap a test-
    // side CRDT replica from it. We'll generate a valid op against
    // this state and send it as A.
    let (buffer_id, snapshot_bytes) =
        match read_message::<InstanceMessage>(&mut stream_a).expect("A first frame") {
            InstanceMessage::BufferSnapshot {
                buffer_id,
                crdt_snapshot,
            } => (buffer_id, crdt_snapshot),
            other => panic!("expected BufferSnapshot, got {other:?}"),
        };

    let a_replica = pmacs::crdt::CrdtState::new(hello_a.assigned_frontend_id.0).expect("a replica");
    a_replica.import_snapshot(&snapshot_bytes).expect("import");

    // A generates a CrdtOp (simulating optimistic-apply at keystroke).
    let v_before = a_replica.version();
    a_replica.insert(0, "X").expect("a insert");
    let op_bytes = a_replica
        .export_updates_since(&v_before)
        .expect("export updates");

    // Drain B's first frame (BufferSnapshot — same buffer_id; the
    // daemon snapshots all CRDT buffers for each replica at attach).
    let _b_first: InstanceMessage = read_message(&mut stream_b).expect("B first frame");

    // A sends FrontendEvent::CrdtOp upstream.
    let crdt_op_event = FrontendEvent::CrdtOp {
        frontend_id: hello_a.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_a.assigned_frontend_id.0,
            bytes: op_bytes.clone(),
        },
    };
    write_message(&mut stream_a, &crdt_op_event).expect("send CrdtOp from A");

    // B should receive an InstanceMessage::CrdtOp tagged with A's
    // frontend_id (via op.peer_id since the wire variant doesn't
    // carry a separate source field).
    stream_b
        .set_read_timeout(Some(Duration::from_millis(500)))
        .unwrap();
    let deadline = Instant::now() + Duration::from_secs(2);
    let mut saw_crdt_op_from_a = false;
    while Instant::now() < deadline {
        match read_message::<InstanceMessage>(&mut stream_b) {
            Ok(InstanceMessage::CrdtOp { buffer_id: bid, op })
                if bid == buffer_id && op.peer_id == hello_a.assigned_frontend_id.0 =>
            {
                saw_crdt_op_from_a = true;
                assert_eq!(
                    op.bytes, op_bytes,
                    "broadcast bytes must match A's emitted bytes"
                );
                break;
            }
            Ok(_) | Err(_) => {}
        }
    }
    assert!(
        saw_crdt_op_from_a,
        "M10.10 Finding 3 criterion: B must receive CrdtOp from A via daemon broadcast"
    );
}

/// M10.10 (post-audit Finding 6) — verify the **production**
/// `attach::build_capabilities()` output negotiates `crdt_replica`
/// correctly so the production TUI binary actually receives
/// `BufferSnapshot` and bootstraps its `BufferMirror`.
///
/// Pre-fix: the M10.10 acceptance tests used `attach_multi()` with
/// custom caps that had `crdt_replica: true`; the production
/// `build_capabilities()` had `crdt_replica: false` so the optimistic-
/// apply infrastructure was structurally unreachable in the real
/// binary. Test coverage didn't catch this because no test used the
/// production caps.
///
/// This test closes that gap: spawn a daemon, connect, send an
/// `AttachRequest` with caps from production `build_capabilities()`,
/// verify the first non-handshake frame is `BufferSnapshot`. If it's
/// `CellDelta` instead, the production capability negotiation is
/// broken.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_production_attach_negotiates_crdt_replica() {
    let daemon = TestDaemon::spawn();
    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();

    // Production handshake — NOT the test `attach_multi()` path.
    let hello: Hello = read_message(&mut stream).expect("read Hello");
    assert_eq!(hello.protocol_version, PROTOCOL_VERSION);
    let req = AttachRequest {
        protocol_version: PROTOCOL_VERSION,
        frontend_capabilities: pmacs::attach::build_capabilities(),
        initial_size: CellSize::new(24, 80),
    };
    write_message(&mut stream, &req).expect("write AttachRequest");

    // Production caps in CRDT-feature builds set `crdt_replica: true`.
    // First frame after the handshake must be `BufferSnapshot`.
    let first: InstanceMessage = read_message(&mut stream).expect("first frame");
    match first {
        InstanceMessage::BufferSnapshot { .. } => {}
        InstanceMessage::CellDelta { .. } => panic!(
            "Finding 6: production frontend received CellDelta as first frame; \
             expected BufferSnapshot. The production caps don't negotiate \
             crdt_replica, so M10.10 optimistic apply is dead in the real TUI."
        ),
        other => {
            panic!("expected BufferSnapshot as first production-frontend frame, got {other:?}")
        }
    }
}

/// M10.10 (post-audit Finding 6 companion) — non-CRDT build path.
/// The production `build_capabilities()` advertises `crdt_replica:
/// false` when built without the `crdt` feature, and the daemon
/// (also non-CRDT) advertises `crdt_replica: false`, so neither side
/// negotiates the capability and the first frame is `CellDelta` as
/// in v0.1. Verifies Finding 3 fix is symmetric on both sides.
#[cfg(not(feature = "crdt"))]
#[test]
fn m10_10_production_attach_non_crdt_build_does_not_negotiate_crdt_replica() {
    let daemon = TestDaemon::spawn();
    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let _hello: Hello = read_message(&mut stream).expect("read Hello");
    let req = AttachRequest {
        protocol_version: PROTOCOL_VERSION,
        frontend_capabilities: pmacs::attach::build_capabilities(),
        initial_size: CellSize::new(24, 80),
    };
    write_message(&mut stream, &req).expect("write AttachRequest");

    let first: InstanceMessage = read_message(&mut stream).expect("first frame");
    match first {
        InstanceMessage::CellDelta { .. } => {}
        other => panic!(
            "non-CRDT build expected CellDelta as first frame; got {other:?}. \
             Finding 3 regression?"
        ),
    }
}

/// M10.10 Day 4 — verify `PMACS_INSTANCE_LATENCY_MS` injection.
///
/// Sets the env var to 200ms; spawns a daemon; attaches a frontend;
/// triggers a `CellDelta` emission by sending a key event; verifies
/// the `CellDelta` arrives at least ~150ms later (allowing for
/// scheduling jitter on busy CI).
///
/// This is the load-bearing setup test for the criterion 1
/// verification — without confirming the injection mechanism
/// works, the latency-dependent criterion tests can't trust their
/// timing.
#[test]
fn m10_10_latency_injection_delays_cell_delta() {
    let daemon = TestDaemon::spawn_with_env(&[("PMACS_INSTANCE_LATENCY_MS", "200")]);
    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let hello = do_handshake(&mut stream);

    // Trigger a CellDelta by sending a key event. v0.1-mode
    // frontend (build_default_caps has crdt_replica=false) so the
    // daemon's path is: receive key → apply edit → render → emit
    // CellDelta with injected sleep.
    let key = FrontendEvent::Key(KeyEvent {
        frontend_id: hello.assigned_frontend_id,
        key: Key::Char('x'),
        mods: Modifiers::NONE,
        timestamp_ns: 0,
    });

    let start = Instant::now();
    write_message(&mut stream, &key).expect("send key");

    // Drain incoming messages until we see a CellDelta.
    loop {
        if matches!(
            read_message::<InstanceMessage>(&mut stream).expect("frame"),
            InstanceMessage::CellDelta { .. }
        ) {
            break;
        }
    }
    let elapsed = start.elapsed();

    // Lower bound: at least 150ms (200ms injection minus jitter
    // tolerance). Upper bound: avoid asserting because CI variability
    // can be high; failing-by-exceeding-bound would be a different
    // kind of bug.
    assert!(
        elapsed >= Duration::from_millis(150),
        "PMACS_INSTANCE_LATENCY_MS=200 should delay CellDelta by ~200ms, \
         observed {elapsed:?}"
    );
}

/// M10.10 Day 4 — criterion 1 acceptance: keystroke send is non-
/// blocking regardless of daemon latency.
///
/// Spec criterion 1: "Local edit visible in less than one frame
/// regardless of instance latency."
///
/// Under Path β, the visible local edit is the optimistic Print
/// emitted synchronously by the frontend's keystroke handler. The
/// daemon's `CellDelta` arrives later (delayed by injected latency)
/// but doesn't block the frontend.
///
/// Demonstration: send 10 `CrdtOp` events back-to-back to a daemon
/// with 200ms injected latency. If the frontend blocked on each
/// `CellDelta`, total time would be ≥2s. The actual time should be
/// under 100ms because writes are non-blocking (the dispatcher's
/// sleeps don't affect the writer's send).
///
/// This is the wire-level demonstration of criterion 1 that
/// complements the orchestrator-level unit test
/// `criterion_1_end_of_line_typing_completes_sub_frame_per_keystroke`.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_criterion_1_keystroke_send_non_blocking_at_200ms_latency() {
    let daemon = TestDaemon::spawn_with_env(&[("PMACS_INSTANCE_LATENCY_MS", "200")]);
    let (hello, mut stream) = attach_multi(&daemon);

    // Bootstrap the test replica from the BufferSnapshot.
    let (buffer_id, snap) = match read_message::<InstanceMessage>(&mut stream).expect("snap") {
        InstanceMessage::BufferSnapshot {
            buffer_id,
            crdt_snapshot,
        } => (buffer_id, crdt_snapshot),
        other => panic!("expected BufferSnapshot, got {other:?}"),
    };
    let replica = pmacs::crdt::CrdtState::new(hello.assigned_frontend_id.0).expect("replica");
    replica.import_snapshot(&snap).expect("import");

    // Simulate 10 back-to-back keystrokes, each producing a CrdtOp
    // sent to the daemon. The frontend's orchestrator is in-process
    // here (the test code is the orchestrator); each send is the
    // analog of "after orchestrator returns CrdtOp + optimistic
    // Print, the wire send happens."
    let start = Instant::now();
    for ch in "0123456789".chars() {
        let v_before = replica.version();
        replica.insert(0, &ch.to_string()).expect("local insert");
        let bytes = replica.export_updates_since(&v_before).expect("export");
        let event = FrontendEvent::CrdtOp {
            frontend_id: hello.assigned_frontend_id,
            buffer_id,
            op: pmacs::rope::CrdtOp {
                peer_id: hello.assigned_frontend_id.0,
                bytes,
            },
        };
        write_message(&mut stream, &event).expect("send");
    }
    let elapsed = start.elapsed();

    // If the frontend's sends blocked on the daemon's 200ms-delayed
    // CellDelta replies, total would be ≥ 2s (10 × 200ms). Non-
    // blocking sends complete in microseconds. Upper bound of 100ms
    // catches any synchronous-IO regression while tolerating CI
    // jitter on the per-write socket cost.
    assert!(
        elapsed < Duration::from_millis(100),
        "criterion 1: 10 keystroke sends at 200ms injected latency took \
         {elapsed:?}; expected non-blocking (<100ms). The frontend is \
         blocking on daemon round-trips."
    );
}

/// M10.10 Day 4 — criterion 2 acceptance: no-flicker via byte-
/// equivalent optimistic paint.
///
/// Spec criterion 2: "Confirmation cell delta does not produce
/// visible flicker or correction (the optimistic state matches the
/// confirmed state)."
///
/// Under Path β's end-of-line scope, the optimistic Print emits the
/// typed character at the cursor's column with the terminal's
/// default style. The daemon's `CellDelta` for the same edit carries
/// a `Cell { glyph: Char(c), style: default }` at the same column.
/// If both encode the same character at the same column with the
/// same style, the daemon's `CellDelta` repaints the cell identically
/// → no visible change → no flicker.
///
/// This test verifies the byte-equivalence property: after the
/// frontend sends a `CrdtOp` for inserting 'X' at end-of-line, the
/// daemon's resulting `CellDelta` carries an 'X' cell at the column
/// where the optimistic Print would have written it. The cell's
/// style is default (no overlays/highlighting active in this
/// minimal test setup).
///
/// **Path β scope**: end-of-line typing only. Mid-line typing
/// produces a multi-cell `CellDelta` (shifted cells); under Path β,
/// no optimistic paint exists for that case (orchestrator round-
/// trips), so there's no optimistic state to flicker against.
/// Documented as v0.2+ Path γ work.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_criterion_2_no_flicker_for_end_of_line_optimistic_insert() {
    use pmacs::cell::{Cell, DiffSpan, Glyph, Style};

    let daemon = TestDaemon::spawn();
    let (hello, mut stream) = attach_multi(&daemon);

    // Bootstrap mirror from BufferSnapshot.
    let (buffer_id, snap) = match read_message::<InstanceMessage>(&mut stream).expect("snap") {
        InstanceMessage::BufferSnapshot {
            buffer_id,
            crdt_snapshot,
        } => (buffer_id, crdt_snapshot),
        other => panic!("expected BufferSnapshot, got {other:?}"),
    };
    let replica = pmacs::crdt::CrdtState::new(hello.assigned_frontend_id.0).expect("replica");
    replica.import_snapshot(&snap).expect("import");

    // The *scratch* buffer starts empty; cursor is at byte 0
    // (which is end-of-line for an empty buffer per the Path β
    // predicate). Insert 'X' at position 0 — both daemon and
    // optimistic Print would put 'X' at column 0 of row 0.
    let v_before = replica.version();
    replica.insert(0, "X").expect("local insert");
    let op_bytes = replica.export_updates_since(&v_before).expect("export");

    let event = FrontendEvent::CrdtOp {
        frontend_id: hello.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello.assigned_frontend_id.0,
            bytes: op_bytes,
        },
    };
    write_message(&mut stream, &event).expect("send CrdtOp");

    // Drain incoming until we find a CellDelta containing the 'X'
    // cell.
    stream
        .set_read_timeout(Some(Duration::from_secs(2)))
        .unwrap();
    let deadline = Instant::now() + Duration::from_secs(3);
    let mut found_x_cell: Option<Cell> = None;
    while Instant::now() < deadline && found_x_cell.is_none() {
        match read_message::<InstanceMessage>(&mut stream) {
            Ok(InstanceMessage::CellDelta { spans, .. }) => {
                for DiffSpan { start, cells } in spans {
                    // Look for an 'X' cell at row 0 (the *scratch*
                    // buffer's only line).
                    for (offset, cell) in cells.iter().enumerate() {
                        if start.row == 0
                            && (start.col as usize + offset) == 0
                            && matches!(cell.glyph, Glyph::Char('X'))
                        {
                            found_x_cell = Some(cell.clone());
                            break;
                        }
                    }
                }
            }
            Ok(_) => {}
            Err(_) => break,
        }
    }

    let cell = found_x_cell.expect(
        "criterion 2: daemon's CellDelta should carry 'X' at column 0 of row 0 \
         after the CrdtOp lands",
    );

    // Byte-equivalence check: the cell's style must be Default
    // (matching what an optimistic Print without explicit style
    // emits to the terminal). Any non-default style here would
    // create visible flicker between optimistic Print (no style)
    // and daemon CellDelta paint (styled).
    assert_eq!(
        cell.style,
        Style::default(),
        "criterion 2: cell at cursor must have default style for byte-equivalent \
         no-flicker repaint. Got {:?}",
        cell.style
    );
    assert!(
        matches!(cell.glyph, Glyph::Char('X')),
        "criterion 2: cell glyph must be 'X'; got {:?}",
        cell.glyph
    );
}

/// M10.10 Day 4 — criterion 3 acceptance: two-frontend conflict
/// convergence.
///
/// Spec criterion 3: "Concurrent edit from another frontend that
/// conflicts with the optimistic edit resolves correctly (CRDT
/// convergence handles it; local frontend's view repaints)."
///
/// Three properties under test:
///
/// (a) **CRDT convergence**: both frontends' mirrors agree on final
///     state after both ops have been delivered.
/// (b) **Local frontend's view repaints**: after convergence, the
///     daemon's `CellDelta` carries the converged content (verifiable
///     on the wire — both frontends receive a `CellDelta` after the
///     remote op is integrated daemon-side).
/// (c) **Optimistic `CrdtOp` preserved through convergence**: each
///     frontend's optimistic edit is in the final state; not lost
///     to the conflict resolution. Verified by inspecting the final
///     mirror content for both frontends' characters.
///
/// The test explicitly sends `FrontendEvent::CrdtOp` (not
/// `FrontendEvent::Key`) so the test proves the optimistic-apply
/// pipeline works end-to-end, not just that some keystroke path
/// produced the right result.
#[cfg(feature = "crdt")]
#[allow(clippy::too_many_lines)] // narrative test with explicit assertions per property
#[test]
fn m10_10_criterion_3_two_frontend_conflict_converges() {
    let daemon = TestDaemon::spawn();

    // Two replica frontends.
    let (hello_a, mut stream_a) = attach_multi(&daemon);
    let (hello_b, mut stream_b) = attach_multi(&daemon);

    // Both receive BufferSnapshot for *scratch* (empty buffer).
    let (buffer_id, snap_a) = match read_message::<InstanceMessage>(&mut stream_a).expect("A snap")
    {
        InstanceMessage::BufferSnapshot {
            buffer_id,
            crdt_snapshot,
        } => (buffer_id, crdt_snapshot),
        other => panic!("expected BufferSnapshot, got {other:?}"),
    };
    let snap_b = match read_message::<InstanceMessage>(&mut stream_b).expect("B snap") {
        InstanceMessage::BufferSnapshot { crdt_snapshot, .. } => crdt_snapshot,
        other => panic!("expected BufferSnapshot, got {other:?}"),
    };

    // Bootstrap test-side replicas from the snapshots. These
    // simulate the frontends' BufferMirror state.
    let replica_a = pmacs::crdt::CrdtState::new(hello_a.assigned_frontend_id.0).expect("a state");
    replica_a.import_snapshot(&snap_a).expect("a import");
    let replica_b = pmacs::crdt::CrdtState::new(hello_b.assigned_frontend_id.0).expect("b state");
    replica_b.import_snapshot(&snap_b).expect("b import");

    // CONCURRENT EDITS: A and B each apply an op to their own
    // mirror BEFORE either sees the other's op. This is the
    // canonical conflict scenario.
    //
    // A optimistically inserts 'A' at position 0 (its mirror state
    // before any remote op has arrived).
    let v_before_a = replica_a.version();
    replica_a.insert(0, "A").expect("a optimistic insert");
    let op_a = replica_a.export_updates_since(&v_before_a).expect("a op");
    assert_eq!(replica_a.materialize_string(), "A");

    // B optimistically inserts 'B' at position 0 (its mirror state
    // before any remote op has arrived — concurrent with A).
    let v_before_b = replica_b.version();
    replica_b.insert(0, "B").expect("b optimistic insert");
    let op_b = replica_b.export_updates_since(&v_before_b).expect("b op");
    assert_eq!(replica_b.materialize_string(), "B");

    // Send both ops to the daemon as FrontendEvent::CrdtOp (the
    // exact wire shape the production optimistic-apply orchestrator
    // emits — property (c) load-bearing assertion).
    let event_a = FrontendEvent::CrdtOp {
        frontend_id: hello_a.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_a.assigned_frontend_id.0,
            bytes: op_a.clone(),
        },
    };
    let event_b = FrontendEvent::CrdtOp {
        frontend_id: hello_b.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_b.assigned_frontend_id.0,
            bytes: op_b.clone(),
        },
    };
    write_message(&mut stream_a, &event_a).expect("send op A");
    write_message(&mut stream_b, &event_b).expect("send op B");

    // Both frontends should receive the OTHER frontend's op via
    // daemon broadcast. Drain incoming streams until each has seen
    // the other's CrdtOp.
    stream_a
        .set_read_timeout(Some(Duration::from_millis(500)))
        .unwrap();
    stream_b
        .set_read_timeout(Some(Duration::from_millis(500)))
        .unwrap();
    let deadline = Instant::now() + Duration::from_secs(3);
    let mut a_received_b_op = false;
    let mut b_received_a_op = false;
    while Instant::now() < deadline && !(a_received_b_op && b_received_a_op) {
        if !a_received_b_op {
            if let Ok(InstanceMessage::CrdtOp { op, .. }) =
                read_message::<InstanceMessage>(&mut stream_a)
            {
                if op.peer_id == hello_b.assigned_frontend_id.0 {
                    replica_a
                        .import_updates(&op.bytes)
                        .expect("a import B's op");
                    a_received_b_op = true;
                }
            }
        }
        if !b_received_a_op {
            if let Ok(InstanceMessage::CrdtOp { op, .. }) =
                read_message::<InstanceMessage>(&mut stream_b)
            {
                if op.peer_id == hello_a.assigned_frontend_id.0 {
                    replica_b
                        .import_updates(&op.bytes)
                        .expect("b import A's op");
                    b_received_a_op = true;
                }
            }
        }
    }
    assert!(
        a_received_b_op,
        "criterion 3: A must receive B's CrdtOp via daemon broadcast"
    );
    assert!(
        b_received_a_op,
        "criterion 3: B must receive A's CrdtOp via daemon broadcast"
    );

    // Property (a): CRDT convergence — both mirrors agree on final
    // state. CRDT order-determinism (loro's peer_id-based ordering)
    // ensures the final content is the same on both replicas.
    let a_final = replica_a.materialize_string();
    let b_final = replica_b.materialize_string();
    assert_eq!(
        a_final, b_final,
        "criterion 3 (a) CRDT convergence: A and B must reach identical final \
         state. A={a_final:?} B={b_final:?}"
    );

    // Property (c): both optimistic edits preserved through
    // convergence — final state contains both 'A' and 'B'.
    assert!(
        a_final.contains('A') && a_final.contains('B'),
        "criterion 3 (c) optimistic edits preserved: final state must contain \
         both A and B. Got {a_final:?}"
    );
    assert_eq!(
        a_final.len(),
        2,
        "criterion 3 (c) edits preserved: final state should be exactly two \
         characters (A and B in some order). Got {a_final:?}"
    );

    // Property (b): the daemon's view of the buffer (which drives
    // CellDelta to other frontends and to fresh attaches) also
    // matches the converged state. Verify by attaching a third
    // observer frontend C and inspecting its BufferSnapshot —
    // it must contain both 'A' and 'B' in the same order.
    let (_hello_c, mut stream_c) = attach_multi(&daemon);
    let snap_c = match read_message::<InstanceMessage>(&mut stream_c).expect("C snap") {
        InstanceMessage::BufferSnapshot { crdt_snapshot, .. } => crdt_snapshot,
        other => panic!("expected BufferSnapshot, got {other:?}"),
    };
    let observer = pmacs::crdt::CrdtState::new(99).expect("observer");
    observer.import_snapshot(&snap_c).expect("observer import");
    let daemon_state = observer.materialize_string();
    assert_eq!(
        daemon_state, a_final,
        "criterion 3 (b) daemon view repaints to converged state: a fresh \
         observer's BufferSnapshot must match what A and B converged to. \
         daemon={daemon_state:?} converged={a_final:?}"
    );
}

/// M10.10 Day 3 Finding 3 acceptance: own-CrdtOp-echo NOT sent back
/// to originator. Frontend A sends `CrdtOp`; daemon broadcasts to
/// other replicas (per-frontend sender-exclusion); A does NOT
/// receive its own op back. This is the daemon-side half of the
/// echo-dedup contract: the frontend-side filter is a defense-in-
/// depth but the daemon shouldn't send echoes in the first place.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_crdt_op_originator_does_not_receive_own_broadcast_echo() {
    let daemon = TestDaemon::spawn();
    let (hello_a, mut stream_a) = attach_multi(&daemon);
    // Need a B attached so the daemon's broadcast loop has someone
    // to broadcast to — without recipients the broadcast is a no-op
    // and the test wouldn't distinguish "no echo because no
    // broadcast" from "no echo because sender-exclusion."
    let (_hello_b, _stream_b) = attach_multi(&daemon);

    // Drain A's BufferSnapshot, bootstrap, generate op.
    let (buffer_id, snapshot_bytes) =
        match read_message::<InstanceMessage>(&mut stream_a).expect("A first frame") {
            InstanceMessage::BufferSnapshot {
                buffer_id,
                crdt_snapshot,
            } => (buffer_id, crdt_snapshot),
            other => panic!("expected BufferSnapshot, got {other:?}"),
        };
    let a_replica = pmacs::crdt::CrdtState::new(hello_a.assigned_frontend_id.0).expect("a replica");
    a_replica.import_snapshot(&snapshot_bytes).expect("import");
    let v_before = a_replica.version();
    a_replica.insert(0, "Q").expect("a insert");
    let op_bytes = a_replica.export_updates_since(&v_before).expect("export");

    let crdt_op_event = FrontendEvent::CrdtOp {
        frontend_id: hello_a.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_a.assigned_frontend_id.0,
            bytes: op_bytes,
        },
    };
    write_message(&mut stream_a, &crdt_op_event).expect("send CrdtOp from A");

    // Read A's incoming stream for a short window; assert no
    // CrdtOp arrives. Subsequent CellDelta / CursorByte / etc.
    // messages are fine — we're just checking CrdtOp specifically.
    stream_a
        .set_read_timeout(Some(Duration::from_millis(200)))
        .unwrap();
    let deadline = Instant::now() + Duration::from_millis(1500);
    while Instant::now() < deadline {
        if let Ok(InstanceMessage::CrdtOp { .. }) = read_message::<InstanceMessage>(&mut stream_a) {
            panic!(
                "M10.10 Finding 3: originator received its own CrdtOp \
                 back — daemon sender-exclusion is broken"
            );
        }
    }
}

// ---------------------------------------------------------------------------
// Post-audit round 2 — F11 / F12 / F13 inbound-CrdtOp validation.
//
// Three negative-path tests proving the daemon rejects malformed or
// out-of-scope inbound `FrontendEvent::CrdtOp` events:
//   - F11: a session that didn't negotiate `crdt_replica` can't drive
//          CRDT state by sending the variant.
//   - F12: an op whose `op.peer_id` names another frontend is rejected
//          before apply / broadcast. (Without this, the receiving
//          frontend's attach loop derives the source via
//          `FrontendId(op.peer_id)` and dedup-skips the broadcast,
//          diverging its mirror from daemon state.)
//   - F13: an op targeting a buffer the source isn't actively editing
//          is rejected. (M10.10's local-edit path only emits ops for
//          the active mirror buffer.)
//
// The reject path drops the op and logs to stderr; we observe rejection
// indirectly by sending a follow-up well-formed op from the same
// frontend and asserting only the well-formed op reaches the peer
// frontend's stream as an `InstanceMessage::CrdtOp` broadcast. (A
// stronger black-box assertion than checking stderr.)
// ---------------------------------------------------------------------------

/// Helper for the F11/F12/F13 negative-path tests: bootstrap two
/// attached replica frontends, drain A's initial frame, return the
/// buffer id A is editing and an A-replica `CrdtState` that's
/// already imported A's snapshot. The caller generates op bytes from
/// the replica and sends crafted `FrontendEvent::CrdtOp` payloads.
#[cfg(feature = "crdt")]
fn bootstrap_two_replicas_for_negative_path(
    daemon: &TestDaemon,
) -> (
    Hello,
    UnixStream,
    Hello,
    UnixStream,
    pmacs::buffer::BufferId,
    pmacs::crdt::CrdtState,
) {
    let (hello_a, mut stream_a) = attach_multi(daemon);
    let (hello_b, mut stream_b) = attach_multi(daemon);
    assert_ne!(hello_a.assigned_frontend_id, hello_b.assigned_frontend_id);

    let (buffer_id, snapshot_bytes) =
        match read_message::<InstanceMessage>(&mut stream_a).expect("A first frame") {
            InstanceMessage::BufferSnapshot {
                buffer_id,
                crdt_snapshot,
            } => (buffer_id, crdt_snapshot),
            other => panic!("expected BufferSnapshot from A, got {other:?}"),
        };
    let _b_first: InstanceMessage = read_message(&mut stream_b).expect("B first frame");

    let a_replica = pmacs::crdt::CrdtState::new(hello_a.assigned_frontend_id.0).expect("a replica");
    a_replica.import_snapshot(&snapshot_bytes).expect("import");

    (hello_a, stream_a, hello_b, stream_b, buffer_id, a_replica)
}

/// Drain stream until a `InstanceMessage::CrdtOp` arrives or the
/// deadline elapses. Used in the negative-path tests to observe the
/// daemon's actual broadcast decisions.
#[cfg(feature = "crdt")]
fn wait_for_crdt_op_broadcast(
    stream: &mut UnixStream,
    deadline: Instant,
) -> Option<(pmacs::buffer::BufferId, pmacs::rope::CrdtOp)> {
    stream
        .set_read_timeout(Some(Duration::from_millis(200)))
        .unwrap();
    while Instant::now() < deadline {
        if let Ok(InstanceMessage::CrdtOp { buffer_id, op }) =
            read_message::<InstanceMessage>(stream)
        {
            return Some((buffer_id, op));
        }
    }
    None
}

/// F11: a frontend that did NOT negotiate `crdt_replica` (legacy v0.1
/// caps) cannot mutate daemon state by sending `FrontendEvent::CrdtOp`.
/// The daemon's pre-apply validation drops the op; the peer replica
/// frontend never receives a broadcast for it.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_f11_non_replica_session_crdt_op_is_rejected() {
    let daemon = TestDaemon::spawn();

    // Legacy session (no crdt_replica) connects first. A is the
    // attacker — its caps don't advertise the capability, but it tries
    // to send a CrdtOp variant anyway.
    let mut stream_a = daemon.connect();
    stream_a
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let hello_a = do_handshake(&mut stream_a);
    let _initial_a: InstanceMessage = read_message(&mut stream_a).expect("A initial frame");

    // B is a legitimate replica frontend that should NOT receive an
    // echo of A's rejected op.
    let (_hello_b, mut stream_b) = attach_multi(&daemon);
    let (buffer_id, snapshot_bytes) =
        match read_message::<InstanceMessage>(&mut stream_b).expect("B first frame") {
            InstanceMessage::BufferSnapshot {
                buffer_id,
                crdt_snapshot,
            } => (buffer_id, crdt_snapshot),
            other => panic!("expected BufferSnapshot, got {other:?}"),
        };

    // A bootstraps a CRDT replica out-of-band (it didn't actually
    // negotiate, but it can still craft bytes since CRDT state is
    // public) and generates an op.
    let a_replica = pmacs::crdt::CrdtState::new(hello_a.assigned_frontend_id.0).expect("a replica");
    a_replica.import_snapshot(&snapshot_bytes).expect("import");
    let v_before = a_replica.version();
    a_replica.insert(0, "Z").expect("insert");
    let op_bytes = a_replica.export_updates_since(&v_before).expect("export");

    let event = FrontendEvent::CrdtOp {
        frontend_id: hello_a.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_a.assigned_frontend_id.0,
            bytes: op_bytes,
        },
    };
    write_message(&mut stream_a, &event).expect("write");

    // B's stream must not see the broadcast within a generous window.
    let deadline = Instant::now() + Duration::from_secs(1);
    assert!(
        wait_for_crdt_op_broadcast(&mut stream_b, deadline).is_none(),
        "F11: daemon must drop CrdtOp from a session that didn't negotiate crdt_replica"
    );
}

/// F12: an op whose `op.peer_id` names a different frontend is
/// rejected before apply / broadcast.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_f12_spoofed_op_peer_id_is_rejected() {
    let daemon = TestDaemon::spawn();
    let (hello_a, mut stream_a, hello_b, mut stream_b, buffer_id, a_replica) =
        bootstrap_two_replicas_for_negative_path(&daemon);

    // Craft ONE set of op bytes (so the daemon can causally apply
    // either copy). Send it twice with different identity framing:
    //   - spoofed event: `op.peer_id` set to B's id (should be
    //     rejected by F12 pre-apply).
    //   - well-formed event: `op.peer_id` set to A's id (should
    //     apply + broadcast normally).
    // Loro is idempotent on remote-op import, so reusing the bytes is
    // safe even if both were applied; here only the second is.
    let v_before = a_replica.version();
    a_replica.insert(0, "X").expect("insert");
    let op_bytes = a_replica.export_updates_since(&v_before).expect("export");

    let spoofed = FrontendEvent::CrdtOp {
        frontend_id: hello_a.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_b.assigned_frontend_id.0, // B's peer_id, not A's
            bytes: op_bytes.clone(),
        },
    };
    let well_formed = FrontendEvent::CrdtOp {
        frontend_id: hello_a.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_a.assigned_frontend_id.0,
            bytes: op_bytes.clone(),
        },
    };
    write_message(&mut stream_a, &spoofed).expect("write spoofed");
    write_message(&mut stream_a, &well_formed).expect("write well-formed");

    // B should see exactly one CrdtOp broadcast — the well-formed one,
    // tagged with A's peer_id. (The spoofed op was rejected pre-apply
    // and would otherwise have arrived with op.peer_id == B.0.)
    let deadline = Instant::now() + Duration::from_secs(2);
    let first = wait_for_crdt_op_broadcast(&mut stream_b, deadline)
        .expect("B must receive the well-formed op");
    assert_eq!(first.0, buffer_id);
    assert_eq!(
        first.1.peer_id, hello_a.assigned_frontend_id.0,
        "F12: B's first CrdtOp broadcast must be the well-formed op (op.peer_id = A's id); \
         a B-tagged op slipping through means the spoofed op was not rejected"
    );
    assert_eq!(first.1.bytes, op_bytes, "F12: broadcast bytes must match");
}

/// F13: an op targeting a `buffer_id` that the source is not
/// actively editing is rejected. We construct this by attaching the
/// session, switching its active window to a different buffer (or
/// using a buffer id that doesn't exist), and asserting the
/// well-formed op for the actually-active buffer still goes through
/// while the wrong-buffer op is dropped.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_f13_wrong_buffer_id_is_rejected() {
    let daemon = TestDaemon::spawn();
    let (hello_a, mut stream_a, _hello_b, mut stream_b, buffer_id, a_replica) =
        bootstrap_two_replicas_for_negative_path(&daemon);

    // A fabricates a non-existent buffer id. The daemon's active-
    // window-for-A lookup returns the scratch buffer; the fabricated
    // id doesn't match → reject. BufferId's constructor is crate-
    // internal, so we round-trip via postcard (its wire form is the
    // bare u64 newtype).
    let fake_buffer_id: pmacs::buffer::BufferId =
        postcard::from_bytes(&postcard::to_stdvec(&u64::MAX).unwrap()).expect("rtt");
    assert_ne!(fake_buffer_id, buffer_id);

    // One set of valid op bytes; send twice with different
    // `buffer_id` framing.
    let v_before = a_replica.version();
    a_replica.insert(0, "X").expect("insert");
    let op_bytes = a_replica.export_updates_since(&v_before).expect("export");

    let wrong_buf = FrontendEvent::CrdtOp {
        frontend_id: hello_a.assigned_frontend_id,
        buffer_id: fake_buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_a.assigned_frontend_id.0,
            bytes: op_bytes.clone(),
        },
    };
    let right_buf = FrontendEvent::CrdtOp {
        frontend_id: hello_a.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_a.assigned_frontend_id.0,
            bytes: op_bytes.clone(),
        },
    };
    write_message(&mut stream_a, &wrong_buf).expect("write wrong-buf");
    write_message(&mut stream_a, &right_buf).expect("write right-buf");

    let deadline = Instant::now() + Duration::from_secs(2);
    let first = wait_for_crdt_op_broadcast(&mut stream_b, deadline)
        .expect("B must receive the right-buffer op");
    assert_eq!(
        first.0, buffer_id,
        "F13: B's first CrdtOp broadcast must target the active buffer; \
         a fake-buffer op slipping through means F13 didn't reject it"
    );
    assert_eq!(first.1.bytes, op_bytes);
}

/// F26: the daemon validates that the loro-internal peer
/// attribution inside `op.bytes` matches the authenticated source.
/// A hostile client can set `op.peer_id == authenticated source`
/// while the update bytes themselves were generated under a
/// DIFFERENT loro peer id; the wrapper-only check (F12) would
/// accept this. Recipients then route/dedup by wrapper identity,
/// but CRDT history attributes to the other peer, splitting
/// observable state from CRDT causal metadata.
///
/// Test shape: A bootstraps a normal CRDT replica under peer-id =
/// `A.0`. A also builds a SECONDARY CRDT replica seeded from A's
/// bootstrap snapshot but with `peer_id = 999`. The secondary
/// replica produces an op; A wraps it with the secondary's loro
/// bytes but `op.peer_id = A.0` (so F12 passes). The daemon's F26
/// fork-import detects the mismatch and rejects. B then submits a
/// well-formed op and only B's broadcast reaches the peer.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_f26_spoofed_loro_internal_peer_id_is_rejected() {
    let daemon = TestDaemon::spawn();
    let (hello_a, mut stream_a, _hello_b, mut stream_b, buffer_id, _a_replica) =
        bootstrap_two_replicas_for_negative_path(&daemon);

    // Secondary replica with a DIFFERENT loro peer id than A's
    // assigned frontend id. Seeded from the same content so its
    // exports are causally applicable to the daemon's buffer.
    let secondary_peer_id: u64 = 999;
    assert_ne!(secondary_peer_id, hello_a.assigned_frontend_id.0);
    let secondary = pmacs::crdt::CrdtState::new(secondary_peer_id).expect("secondary");
    let snapshot_donor = pmacs::crdt::CrdtState::new(2).expect("snap-donor");
    let snapshot_bytes = snapshot_donor.export_snapshot().expect("export");
    secondary.import_snapshot(&snapshot_bytes).expect("import");
    let v_before = secondary.version();
    secondary.insert(0, "S").expect("secondary insert");
    let spoofed_bytes = secondary
        .export_updates_since(&v_before)
        .expect("export spoofed");

    // Send the spoofed op: wrapper peer_id matches A (F12 passes)
    // but the bytes carry ops attributed to peer 999. F26 fork-
    // import detects this and rejects.
    let spoofed = FrontendEvent::CrdtOp {
        frontend_id: hello_a.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_a.assigned_frontend_id.0, // wrapper = A (F12 passes)
            bytes: spoofed_bytes.clone(),
        },
    };
    write_message(&mut stream_a, &spoofed).expect("write spoofed");

    // B must not receive a broadcast for the spoofed op within a
    // generous window. (The peer's broadcast would carry op.peer_id
    // = A but the bytes attributed to peer 999 — if accepted, the
    // peer mirror would dedup-skip on echo and then re-apply via
    // the broadcast, getting wrong attribution either way.)
    stream_b
        .set_read_timeout(Some(Duration::from_millis(200)))
        .unwrap();
    let deadline = Instant::now() + Duration::from_secs(1);
    let mut saw_spoofed_broadcast = false;
    while Instant::now() < deadline {
        if let Ok(InstanceMessage::CrdtOp { op, .. }) =
            read_message::<InstanceMessage>(&mut stream_b)
        {
            if op.bytes == spoofed_bytes {
                saw_spoofed_broadcast = true;
                break;
            }
        }
    }
    assert!(
        !saw_spoofed_broadcast,
        "F26: daemon must reject ops whose loro-internal peer attribution \
         disagrees with the authenticated source"
    );
}

/// F16: a daemon-side mutation (a `FrontendEvent::Key` round-trip)
/// that generates a CRDT op broadcasts to **all** replica frontends
/// including the source. Pre-fix, the source's mirror would silently
/// drift from daemon state after every fallback / Key-path edit
/// because `pending_crdt_ops` excluded the active frontend.
///
/// Test shape: A (replica) attaches; sends a plain `FrontendEvent::Key`
/// (a printable char that the daemon's command pipeline maps to
/// `pmacs.editor.insert-text`). The daemon mutates the active buffer,
/// generates a CRDT op tagged `CrdtOpOrigin::DaemonKey`, broadcasts.
/// A asserts it received its own `InstanceMessage::CrdtOp` broadcast.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_f16_daemon_key_origin_broadcasts_to_source() {
    let daemon = TestDaemon::spawn();
    let (hello_a, mut stream_a) = attach_multi(&daemon);

    // Drain A's bootstrap frames (BufferSnapshot + CursorByte).
    // We're going to wait for a CrdtOp specifically, so we want
    // any other messages drained first.
    let _first_a: InstanceMessage =
        read_message(&mut stream_a).expect("A first frame (BufferSnapshot)");

    // Send a `FrontendEvent::Key` for a plain printable char. The
    // daemon's normal key-path will produce a CRDT op tagged
    // `DaemonKey`; F16 ensures it broadcasts to A as well.
    let key_event = FrontendEvent::Key(KeyEvent {
        frontend_id: hello_a.assigned_frontend_id,
        key: Key::Char('K'),
        mods: Modifiers::NONE,
        timestamp_ns: 0,
    });
    write_message(&mut stream_a, &key_event).expect("write Key");

    // A must receive its OWN edit's CRDT broadcast (F16). Without
    // the fix, the broadcast would exclude A and the assertion would
    // time out.
    stream_a
        .set_read_timeout(Some(Duration::from_millis(200)))
        .unwrap();
    let deadline = Instant::now() + Duration::from_secs(2);
    let mut saw_self_broadcast = false;
    while Instant::now() < deadline {
        if let Ok(InstanceMessage::CrdtOp { .. }) = read_message::<InstanceMessage>(&mut stream_a) {
            saw_self_broadcast = true;
            break;
        }
    }
    assert!(
        saw_self_broadcast,
        "F16: daemon-key-origin CRDT op must broadcast to the active frontend; \
         pre-fix the source frontend's mirror diverged from daemon state on every \
         Key round-trip"
    );
}

/// F17: a CRDT op whose import produces no text delta (e.g. a
/// concurrent same-character delete that already converged) is
/// **still broadcast** to peers. Pre-fix, `handle_remote_crdt_op`
/// returned early on `Ok(None)` and dropped the op — peers never
/// imported the CRDT causal metadata, and later updates that
/// depend on it could fail to apply.
///
/// Test shape: A and B both attached as replicas. A bootstraps a
/// CRDT replica and generates an op. A sends the op twice. The
/// first application produces an Edit and broadcasts; the second
/// `apply_remote_crdt_op` is idempotent → `Ok(None)`. F17 ensures
/// the second broadcast still happens, so B sees **two** `CrdtOp`
/// broadcasts even though the daemon's text content changed only
/// once.
#[cfg(feature = "crdt")]
#[test]
fn m10_10_f17_no_text_delta_import_still_broadcasts() {
    let daemon = TestDaemon::spawn();
    let (hello_a, mut stream_a, _hello_b, mut stream_b, buffer_id, a_replica) =
        bootstrap_two_replicas_for_negative_path(&daemon);

    let v_before = a_replica.version();
    a_replica.insert(0, "Z").expect("insert");
    let op_bytes = a_replica.export_updates_since(&v_before).expect("export");

    let event = FrontendEvent::CrdtOp {
        frontend_id: hello_a.assigned_frontend_id,
        buffer_id,
        op: pmacs::rope::CrdtOp {
            peer_id: hello_a.assigned_frontend_id.0,
            bytes: op_bytes.clone(),
        },
    };
    // First send: applies (Some(edit)) and broadcasts.
    write_message(&mut stream_a, &event).expect("write 1");
    // Second send: import is idempotent → Ok(None). Pre-fix the op
    // would have been dropped here; F17 still pushes to the broadcast
    // queue.
    write_message(&mut stream_a, &event).expect("write 2");

    // B must see TWO CrdtOp broadcasts, both carrying the same
    // bytes. Without F17, B would see only one.
    stream_b
        .set_read_timeout(Some(Duration::from_millis(200)))
        .unwrap();
    let deadline = Instant::now() + Duration::from_secs(2);
    let mut crdt_op_count = 0;
    while Instant::now() < deadline && crdt_op_count < 2 {
        if let Ok(InstanceMessage::CrdtOp { op, .. }) =
            read_message::<InstanceMessage>(&mut stream_b)
        {
            assert_eq!(op.bytes, op_bytes);
            crdt_op_count += 1;
        }
    }
    assert_eq!(
        crdt_op_count, 2,
        "F17: peer must receive both broadcasts even though the second \
         apply_remote_crdt_op produced no text delta; pre-fix only one arrives"
    );
}

/// F14: production-path end-to-end keystroke flow. Negotiates with
/// the **production** `attach::build_capabilities()` (not the test
/// `multi_frontend_caps()`), bootstraps a real `BufferMirror` from
/// the daemon's `BufferSnapshot` + `CursorByte`, then drives the
/// production `optimistic::frontend_event_for_keystroke` orchestrator
/// with a synthetic `KeyEvent`. The produced `FrontendEvent::CrdtOp`
/// is sent to the daemon and a second replica frontend (B) must
/// receive the broadcast.
///
/// Why this matters: the other M10.10 acceptance tests inject
/// `FrontendEvent::CrdtOp` directly with hand-built fields. That
/// bypasses the production decision chain (`classify_key`,
/// `frontend_event_for_keystroke`, eligibility predicates), so a
/// regression in *any* of those layers would not be caught by the
/// existing matrix — exactly the gap that allowed the original
/// post-audit Finding 1 to ship a structurally-unreachable
/// optimistic-apply path.
///
/// We can't drive crossterm's raw-mode terminal from a test, but we
/// can drive everything from `KeyEvent` downward in the same code
/// path the production attach loop uses (see src/attach.rs:784).
#[cfg(feature = "crdt")]
#[test]
fn m10_10_f14_production_path_keystroke_flows_to_broadcast() {
    use pmacs::buffer_mirror::BufferMirror;

    let daemon = TestDaemon::spawn();

    // A attaches with PRODUCTION caps — exercises Finding 3 fix
    // (build_capabilities advertising crdt_replica in CRDT builds).
    let mut stream_a = daemon.connect();
    stream_a
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let hello_a: Hello = read_message(&mut stream_a).expect("A Hello");
    let req_a = AttachRequest {
        protocol_version: PROTOCOL_VERSION,
        frontend_capabilities: pmacs::attach::build_capabilities(),
        initial_size: CellSize::new(24, 80),
    };
    write_message(&mut stream_a, &req_a).expect("A AttachRequest");

    // B attaches with multi-frontend test caps so it's a replica too
    // — it will receive A's broadcast.
    let (_hello_b, mut stream_b) = attach_multi(&daemon);

    // A's bootstrap: BufferSnapshot first, then drain frames until
    // CursorByte arrives (the byte-position pairing M10.10 added).
    let (buffer_id, snapshot_bytes) = match read_message::<InstanceMessage>(&mut stream_a)
        .expect("A first frame")
    {
        InstanceMessage::BufferSnapshot {
            buffer_id,
            crdt_snapshot,
        } => (buffer_id, crdt_snapshot),
        other => panic!(
            "F14: production caps must negotiate crdt_replica → first frame is BufferSnapshot; got {other:?}"
        ),
    };

    let mut a_mirror = BufferMirror::new(hello_a.assigned_frontend_id);
    a_mirror
        .init_from_snapshot(buffer_id, &snapshot_bytes)
        .expect("init_from_snapshot");
    a_mirror.set_cursor_byte_pos(buffer_id, 0);

    // Drain until CursorByte arrives (or timeout) so the mirror's
    // cursor is grounded in the daemon's actual cursor position. The
    // attach loop normally does this; we replicate it here.
    let drain_deadline = Instant::now() + Duration::from_millis(500);
    stream_a
        .set_read_timeout(Some(Duration::from_millis(100)))
        .unwrap();
    while Instant::now() < drain_deadline {
        match read_message::<InstanceMessage>(&mut stream_a) {
            Ok(InstanceMessage::CursorByte {
                buffer_id: bid,
                byte_pos,
            }) if bid == buffer_id => {
                a_mirror.set_cursor_byte_pos(buffer_id, byte_pos as usize);
                break;
            }
            Ok(_) | Err(_) => {}
        }
    }

    // Drive the PRODUCTION orchestrator. This is the exact call site
    // at src/attach.rs:784 — same function signature, same inputs.
    let pmacs_key = KeyEvent {
        frontend_id: hello_a.assigned_frontend_id,
        key: Key::Char('Q'),
        mods: Modifiers::NONE,
        timestamp_ns: 0,
    };
    let frontend_event = pmacs::optimistic::frontend_event_for_keystroke(
        &mut a_mirror,
        hello_a.assigned_frontend_id,
        pmacs_key,
    );

    // The orchestrator must produce a CrdtOp (mirror is ready, action
    // is Insert, cursor is at a valid position). If it returned
    // `Key` as fallback, the production path's optimistic apply is
    // broken end-to-end.
    let (event_buffer_id, op_peer_id, op_bytes) = match &frontend_event {
        FrontendEvent::CrdtOp { buffer_id, op, .. } => (*buffer_id, op.peer_id, op.bytes.clone()),
        other => panic!(
            "F14: production orchestrator returned {other:?} instead of CrdtOp \
             for a plain Char('Q') insert. This means the mirror wasn't ready or \
             the eligibility predicates rejected what should be a viable optimistic \
             insert. (See src/optimistic.rs:128 — `frontend_event_for_keystroke`.)"
        ),
    };
    assert_eq!(event_buffer_id, buffer_id);
    assert_eq!(op_peer_id, hello_a.assigned_frontend_id.0);

    // Send the produced event upstream; B must see the broadcast.
    write_message(&mut stream_a, &frontend_event).expect("write CrdtOp");
    let deadline = Instant::now() + Duration::from_secs(2);
    let broadcast = wait_for_crdt_op_broadcast(&mut stream_b, deadline)
        .expect("F14: B must receive A's CrdtOp broadcast end-to-end");
    assert_eq!(broadcast.0, buffer_id);
    assert_eq!(broadcast.1.peer_id, hello_a.assigned_frontend_id.0);
    assert_eq!(
        broadcast.1.bytes, op_bytes,
        "F14: broadcast bytes must match what the production orchestrator produced"
    );
}

/// M10.10 Day 3 Finding 2 acceptance: a non-replica frontend does NOT
/// receive `CursorByte` — capability gating skips the emission for
/// frontends that don't negotiate `crdt_replica`. (Without this gate,
/// the non-replica frontend's postcard decoder would hard-error on
/// the unknown variant per Refinement 3.)
#[test]
fn m10_10_non_replica_frontend_does_not_receive_cursor_byte() {
    let daemon = TestDaemon::spawn();
    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_millis(500)))
        .unwrap();
    let _hello = do_handshake(&mut stream);

    // Read a handful of incoming frames; assert none is CursorByte.
    // 16 frames is enough to cover at least a couple of per-tick
    // render cycles.
    for _ in 0..16 {
        match read_message::<InstanceMessage>(&mut stream) {
            Ok(InstanceMessage::CursorByte { .. }) => panic!(
                "non-replica frontend received CursorByte — capability \
                 gating must skip the emission"
            ),
            Ok(_) => {}
            Err(_) => break,
        }
    }
}

/// M10.10 Day 2 acceptance: a frontend that did NOT negotiate
/// `crdt_replica` (the v0.1-style frontend path) receives the initial
/// `CellDelta` directly without any `BufferSnapshot` first. The
/// daemon's M10.10 send is capability-gated; non-replica frontends
/// don't even see the wire variant (and can't decode it — postcard
/// hard-errors on unknown variants per Refinement 3).
#[test]
fn m10_10_non_replica_frontend_does_not_receive_buffer_snapshot() {
    let daemon = TestDaemon::spawn();
    let mut stream = daemon.connect();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    // build_default_caps() has crdt_replica: false — the daemon
    // negotiates crdt_replica: false and skips the send.
    let _hello = do_handshake(&mut stream);

    let first: InstanceMessage = read_message(&mut stream).expect("first frame");
    match first {
        InstanceMessage::CellDelta {
            full_grid: true, ..
        } => {}
        InstanceMessage::BufferSnapshot { .. } => {
            panic!(
                "M10.10: non-replica frontend received BufferSnapshot — \
                 capability gating must skip the send"
            );
        }
        other => panic!(
            "M10.10: first frame for non-replica frontend should be full-grid \
             CellDelta, got {other:?}"
        ),
    }
}