koh 0.12.0

//! In-process harnesses that exercise the full koh stack *without* iroh or a real terminal, by
//! wiring the client and server transports through the deterministic chaotic link in
//! [`crate::ssp::testkit`]. This is the integration + chaos coverage: it proves the screen and
//! input states converge end-to-end under loss, that the predictor confirms/suppresses against
//! real frames, and (via the harness's own guard) that superseded screens are never delivered
//! late. Driven by `tests/integration.rs` (automated) and the `chaos` example (manual).

use crate::input::{UserInput, WireEvent};
use crate::predict::{DisplayPreference, PredictionEngine};
use crate::ssp::testkit::{GridState, LinkParams, SimHarness};
use crate::terminal::{ServerTerminal, TerminalScreen};

/// The outcome of one driven client↔server session over the chaotic link.
pub struct SessionResult {
    pub converge_steps: usize,
    pub sim_ms: u64,
    pub client_text: String,
    pub client_echo_ack: u64,
    pub expected_frame: u64,
}

impl SessionResult {
    pub fn assert_ok(&self) -> anyhow::Result<()> {
        if !self.client_text.contains("hello koh") {
            anyhow::bail!(
                "client screen missing expected output; got:\n{}",
                self.client_text
            );
        }
        if self.client_echo_ack != self.expected_frame {
            anyhow::bail!(
                "echo_ack mismatch: client={} expected={}",
                self.client_echo_ack,
                self.expected_frame
            );
        }
        Ok(())
    }
}

/// Drive a full client↔server session over the lossy link to convergence.
///
/// Wires client (A: `Transport<UserInput, TerminalScreen>`) to server
/// (B: `Transport<TerminalScreen, UserInput>`), types a command, has a fake shell echo it, and
/// drives everything to convergence — including the server's echo-ack.
pub fn run_session(loss: f64, seed: u64) -> SessionResult {
    let params = LinkParams {
        loss,
        min_delay_ms: 10,
        max_delay_ms: 60,
        dup: 0.02,
    };
    let mut h = SimHarness::<UserInput, TerminalScreen>::new(params, seed, 1200);

    // The server's authoritative emulator with an initial prompt.
    let mut emu = ServerTerminal::new(24, 80, 0);
    emu.process(b"$ ");
    *h.b_mut() = emu.snapshot();

    // Initial sync: client receives the prompt.
    h.run_until(5_000, |h| {
        h.a.remote_state().screen().contents().contains('$')
    });

    // Client types a command (with the trailing CR a shell would see).
    let cmd = b"echo hello koh\r";
    h.a_mut().push_bytes(cmd);

    // Drive until the server has received the whole command.
    h.run_until(20_000, |h| h.b.remote_state().events().len() >= cmd.len());

    // The fake shell: drain the input, echo each byte, and on CR emit the command output.
    let frame = h.b.remote_num();
    let arrival = h.now();
    // The per-connection echo-ack tracker the server loop owns (KS-02).
    let mut echo = crate::server::EchoAck::default();
    echo.register_input_frame(frame, arrival);
    for w in h.b.get_remote_diff() {
        if let WireEvent::Keys(bytes) = w {
            for b in bytes {
                if b == b'\r' {
                    emu.process(b"\r\nhello koh\r\n$ ");
                } else {
                    emu.process(&[b]);
                }
            }
        }
    }
    // Past the echo-ack debounce: the input is now reflected on screen. Stamp the ack the way the
    // connection loop does, after taking the snapshot.
    echo.set_echo_ack(arrival + 1_000);
    let mut snap = emu.snapshot();
    snap.set_echo_ack(echo.echo_ack());
    *h.b_mut() = snap.clone();

    let target = snap;
    let converge_steps = h.run_until(40_000, |h| *h.a.remote_state() == target);

    SessionResult {
        converge_steps,
        sim_ms: h.now(),
        client_text: h.a.remote_state().screen().contents(),
        client_echo_ack: h.a.remote_state().echo_ack(),
        expected_frame: frame,
    }
}

/// Drive a client-side predictor against authoritative frames to confirm the epoch gate:
/// a keystroke stays hidden until the server confirms it echoes, then is confirmed-and-cleared.
pub fn run_predictor_reconciliation() -> anyhow::Result<()> {
    let mut pe = PredictionEngine::new(DisplayPreference::Always);
    pe.set_local_frame_sent(0);

    // SECURITY: the first keystroke is epoch-gated and must NOT be shown before confirmation.
    let blank = TerminalScreen::default();
    pe.new_user_byte(100, b'h', blank.screen());
    if !pe.overlay(blank.screen()).is_empty() {
        anyhow::bail!("predictor leaked a keystroke before the server confirmed it echoes");
    }

    // Server echoes 'h' and acks frame 1 -> prediction confirmed and cleared.
    let echoed = TerminalScreen::from_bytes(24, 80, b"h");
    pe.set_local_frame_late_acked(1);
    pe.cull(200, echoed.screen());
    if !pe.overlay(echoed.screen()).is_empty() {
        anyhow::bail!("confirmed prediction should be cleared");
    }
    Ok(())
}

/// The outcome of one driven session over a **non-terminal** state (KH-01).
///
/// A scripted producer mutates [`GridState`] cells on the server side while the client types; both
/// directions must converge over the chaotic link exactly as the terminal session does.
pub struct GenericSessionResult {
    pub converge_steps: usize,
    pub sim_ms: u64,
    pub client_view: GridState,
    pub server_view_of_input: usize,
    pub expected_input: usize,
}

impl GenericSessionResult {
    pub fn assert_ok(&self, expected: &GridState) -> anyhow::Result<()> {
        if self.client_view != *expected {
            anyhow::bail!(
                "client replica diverged from the server state: {} cells vs {}",
                self.client_view.cells.len(),
                expected.cells.len()
            );
        }
        if self.server_view_of_input != self.expected_input {
            anyhow::bail!(
                "server saw {} input events, expected {}",
                self.server_view_of_input,
                self.expected_input
            );
        }
        Ok(())
    }
}

/// Drive a client↔server session whose synced state is a [`GridState`] over the lossy link (KH-01).
///
/// A stand-in for a multiplexer's per-pane grids: `rounds` rounds of random cell mutations (some
/// larger than one datagram) interleaved with client keystrokes, then convergence. Returns the
/// result and the final server state to compare against.
pub fn run_generic_session(loss: f64, seed: u64, rounds: u32) -> (GenericSessionResult, GridState) {
    let params = LinkParams {
        loss,
        min_delay_ms: 10,
        max_delay_ms: 60,
        dup: 0.02,
    };
    let mut h = SimHarness::<UserInput, GridState>::new(params, seed, 1200);
    let mut rng = crate::ssp::testkit::Rng::new(seed ^ 0xA5A5);
    let mut typed = 0usize;
    for round in 0..rounds {
        let k = (rng.next_u64() % 12) as u32;
        let len = rng.range(1, 2500) as usize;
        h.b_mut().cells.insert(k, vec![round as u8; len]);
        if (rng.next_u64()).is_multiple_of(3) {
            h.b_mut().cells.remove(&((rng.next_u64() % 12) as u32));
        }
        h.a_mut().push_bytes(b"k");
        typed += 1;
        h.run_steps(5);
    }
    let target = h.b.current().clone();
    let converge_steps = h.run_until(60_000, |h| {
        *h.a.remote_state() == target && h.b.remote_state().events().len() >= typed
    });
    let result = GenericSessionResult {
        converge_steps,
        sim_ms: h.now(),
        client_view: h.a.remote_state().clone(),
        server_view_of_input: h.b.remote_state().events().len(),
        expected_input: typed,
    };
    (result, target)
}