diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index 112ba67f..b16f6472 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -177,10 +177,12 @@ jobs: engine bash /tmp/e2e/scripts/e2e_stack_smoke.sh http://localhost:8910 # The live Playwright project, in-network. A Playwright - # container on the compose network runs the SAME gui.spec.ts specs against + # container on the compose network runs the `live`-project specs against # http://engine:8910 with the real API (no installFixtures), so mock/truth - # drift fails a test. The runner cannot reach the stack over host loopback, - # so this runs via `docker compose run` on the network like the smoke walk. + # drift fails a test. This includes gui.spec.ts and terminalReplay.spec.ts + # (H3, WI-124: the terminal-attach-budget symptoms against a real load + # session). The runner cannot reach the stack over host loopback, so this + # runs via `docker compose run` on the network like the smoke walk. # # First landing is advisory (continue-on-error): the in-network smoke walk # above is the hard end-to-end gate, and this project is being brought up diff --git a/docs/ENGINE.md b/docs/ENGINE.md index 7a39614a..068be6b1 100644 --- a/docs/ENGINE.md +++ b/docs/ENGINE.md @@ -623,8 +623,9 @@ token can list sessions and read a scrollback snapshot over The attach sequence is ordered: -1. client sends auth control frame, optionally with its last applied cursor: - `{"type":"auth","token":"...","resume_from":123}` +1. client sends auth control frame, optionally with its last applied cursor + and always with its replay budget: + `{"type":"auth","token":"...","resume_from":123,"snapshot_tail_bytes":1048576}` 2. server sends `snapshot-start` 3. server sends zero or more binary scrollback chunks 4. server sends `snapshot-done` @@ -634,7 +635,7 @@ Client text control frames: ```json {"type":"resize","cols":120,"rows":40} -{"type":"ping"} +{"type":"ping","id":1} ``` Server text control frames: @@ -642,12 +643,37 @@ Server text control frames: ```json {"type":"snapshot-start","session_id":"uuid","scrollback_bytes":0,"scrollback_pos":0,"reset":true} {"type":"snapshot-done"} +{"type":"pong","id":1,"pos":123} {"type":"lag","note":"client too slow; reattach"} ``` -When `resume_from` is still retained, `snapshot-start.reset` is false and the -binary snapshot contains only newer bytes. Otherwise the server returns a full -snapshot with `reset` true. +A `pong` echoes the `ping.id` and carries `pos`: the absolute byte offset the +server has **actually streamed to that socket**, not `total_written`. The pong +is formed and sent by the same outbound task that streams output, after +flushing anything already queued, so it is ordered on the wire after those +chunks and its `pos` can never exceed what the client has received. A client +uses it purely as a liveness probe — a `pos` ahead of what it has rendered is a +suspect to confirm with a second probe, not an immediate reconnect. + +`snapshot_tail_bytes` is the replay **budget**, and it bounds every reply — a +client's terminal keeps only a fixed scrollback, so the server never ships more +than the client can hold: + +- **Cold attach** (no `resume_from`): a ground-state-aligned tail of at most the + budget, `reset` true. +- **Warm reattach whose cursor is retained and whose delta fits the budget:** + `reset` false and the binary snapshot contains only the newer bytes, + byte-for-byte — an ordinary switch-away/switch-back appends without a clear. +- **Warm reattach whose cursor aged out of the ring, or whose delta exceeds the + budget:** a ground-state-aligned tail of at most the budget, `reset` true. + `reset: true` may therefore follow a `resume_from`. The client discards its + stale cursor and re-anchors to `scrollback_pos - scrollback_bytes` (the start + of the tail), so after replaying the tail its position is `scrollback_pos` + again and live traffic resumes with no gap. + +The returned `scrollback_pos` is always the absolute end position, unaffected by +trimming the front. Omitting `snapshot_tail_bytes` (the in-band lag resync, +which carries its own cursor) leaves a retained delta unbounded and byte-exact. A text frame that does not parse as a control message is treated as raw input, because some tools send keystrokes as text. Snapshot chunks are capped at diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md new file mode 100644 index 00000000..022d40cc --- /dev/null +++ b/docs/ROADMAP.md @@ -0,0 +1,111 @@ +# Roadmap and design notes + +Forward-looking design decisions that are not yet work in flight. Each entry is +a recommendation with enough reasoning and numbers to act on — or to decide +"not now" without re-deriving the analysis. + +## Engine-side headless VT per session (D6, WI-130) + +*Spike, 2026-09-09. Part of the terminal-attach-budget initiative (WI-121).* + +### Question + +The engine keeps only **raw PTY bytes** in a per-session ring +(`scrollback.rs`), and every client rebuilds a terminal by re-parsing those +bytes into xterm.js on attach. Should the engine instead run a **headless +terminal emulator per session**, so a cold attach ships *screen state* +(tmux-style) — a serialized screen plus N lines of scrollback — instead of byte +history? + +### What the budget work already fixed + +Before committing to a VT, weigh it against what shipped in this initiative: + +- **F1 (WI-125)** bounds *every* replay to the tail budget — the 4 MiB aged-out + flood is gone; a cold or stale attach now ships at most the budget + (default 1 MiB), ground-state aligned. +- **F3 (WI-127)** stops retained tabs from time-slicing the active pane's + parser on reload. +- **F5 (WI-129)** persists a serialized xterm screen client-side, so a *reload* + restores in one write with no raw re-parse, and (once F4 lands) a *switch* + keeps the socket open and never re-streams. + +So the two operator symptoms (switch re-stream, slow fresh open) are addressed +by bounding and by client-side serialization, **without** an engine VT. The VT +is only worth its cost if a residual problem remains after these ship and are +measured on prod — see "When to revisit". + +### Crate options + +| Crate | What it is | Fit for a screen-snapshot server VT | +|---|---|---| +| `vt100` | A minimal pure-Rust parser that maintains a screen grid + scrollback, no rendering. | **Best fit.** Smallest surface and memory; exposes the cell grid directly, which is all a snapshot needs. Serializing its screen to an xterm-compatible escape stream is a bounded amount of new code. | +| `alacritty_terminal` | The terminal model behind Alacritty. | Heavier per-cell model and an API shaped around a GPU renderer's needs; more memory and more moving parts than a snapshot server wants, and version churn tied to Alacritty. | +| `termwiz` | WezTerm's terminal library. | Full-featured (its own line/cell/attribute model, image protocols); the largest surface of the three. Overkill for "hold a screen and emit a snapshot". | + +### Cost estimate (per session, at 8 busy sessions) + +No live measurement was taken (an agent session cannot reach a prod stack; use +H1's load session — `load_session_command` in the engine integration tests — to +measure before adopting). Grounded estimate for `vt100`: + +- **Memory.** A cell is a codepoint + attributes ≈ 8–16 B. A 200-col grid with + 50 visible rows + 1000 scrollback lines ≈ 210 000 cells ≈ **1.7–3.4 MB per + session**, i.e. **~14–27 MB across 8 sessions** — *on top of* the existing + 4 MiB raw ring per session unless the ring is then shrunk. The ring cannot be + dropped entirely: warm `resume_from` deltas and the history archive still read + raw bytes. +- **CPU.** The VT must parse every byte the PTY produces, on the hot reader + path. The client already parses at ~5–6 MB/s into xterm; a Rust grid VT is + faster (tens of MB/s) but it is now paid **once per session, always**, not + once per attach. For a chatty agent session this is continuous cost the + raw-ring design does not have. + +### Wire shape (if adopted) + +Add a snapshot kind rather than changing the existing one: + +```json +{"type":"snapshot-start","kind":"screen","session_id":"…","scrollback_pos":N,"reset":true} +``` + +followed by a server-produced, xterm-compatible escape stream that redraws the +screen and the last N scrollback lines, then live bytes as today. `resume_from` +still selects the byte-delta path (unchanged); `kind:"screen"` is the cold / +aged-out path only. The history archive is untouched (it keeps raw bytes). + +### Fidelity + +A server VT and the client xterm are two independent emulators; they can +disagree (edge cases in wide chars, unusual SGR, DEC private modes). The +snapshot is only correct if the server's redraw reproduces on xterm what the +program intended. The H2 corpus + fidelity harness is the right place to prove +this: parse each corpus in the candidate crate, emit the screen escape stream, +write it into headless xterm, and compare `serialize()` against xterm fed the +raw bytes. Adopt only if that comparison is clean across the corpus. + +### Mobile + +The prize on mobile would be dropping the IndexedDB cache entirely: if a cold +attach always ships a small screen snapshot, the phone need not persist raw +scrollback. But F5 already gives mobile a one-write restore from a *bounded* +serialized cache, so the marginal win is "no client cache at all" versus "a +small, capped client cache" — modest. + +### Recommendation: **not now** + +Ship and measure F1–F5 (and F4) on prod first. The engine VT doubles per-session +memory and adds continuous CPU on the PTY hot path to solve a problem the +bounding + client serialization work already targets. Revisit **only** if, with +those shipped and measured: + +1. a first cold attach on mobile is still too slow *and* the ≤1 MiB tail is the + bottleneck (not network), or +2. keeping any client-side scrollback cache proves untenable (storage, privacy), + or +3. a new requirement wants server-authoritative screen state (e.g. server-side + search over live screens, or thumbnails). + +If revisited, the first step is a `vt100`-based measurement against H1's load +session at 8 sessions, then the H2 fidelity comparison above — no new work items +are filed now. diff --git a/engine/server/src/pty.rs b/engine/server/src/pty.rs index 8d28cbcf..e0c0133a 100644 --- a/engine/server/src/pty.rs +++ b/engine/server/src/pty.rs @@ -130,13 +130,25 @@ impl Session { /// Snapshot only output newer than a client cursor. The boolean tells the /// client whether its existing terminal state must be reset. /// - /// `tail_bytes` caps a **cold** attach only: a fresh client with no - /// cache sends no `resume_from`, and without a cap the full snapshot is the - /// entire scrollback ring. When present it bounds that full snapshot to at - /// most that many trailing bytes. A warm reattach (`resume_from` present) - /// ignores it entirely, so its `reset: false` delta stays byte-for-byte - /// unchanged; the returned position (`total_written`) is unaffected by - /// trimming the front, so the live stream still resumes with no gap. + /// `tail_bytes` is the replay **budget**, and it now bounds *every* path, + /// not just a cold attach (F1, WI-125). A client's xterm keeps only a fixed + /// scrollback (5000 lines); shipping more than the budget is wasted parse + /// on the client's single message thread, and on a stale reattach it is + /// parsed *above* a screen the client already has out of date. So: + /// + /// - Cold attach (no `resume_from`): bound the full snapshot to the budget. + /// - Warm reattach whose cursor is still retained and whose delta fits the + /// budget: the delta is sent byte-for-byte with `reset: false`, so a + /// switch-away/switch-back that produced little still appends cleanly. + /// - Warm reattach whose cursor aged out of the ring, **or** whose delta + /// exceeds the budget: a ground-state-aligned tail of at most the budget, + /// with `reset: true`. Never the whole untrimmed ring — that was the + /// 4 MiB flood, 4x worse than a cold attach. + /// + /// The returned position is always `total_written`, unaffected by trimming + /// the front, so the live stream resumes with no gap. When `tail_bytes` is + /// `None` (the in-band resync path, which carries its own cursor) nothing + /// is bounded and a retained delta stays byte-exact. pub fn snapshot_for_attach( &self, resume_from: Option, @@ -146,12 +158,25 @@ impl Session { let pos = sb.total_written(); if let Some(cursor) = resume_from { if let Some(delta) = sb.snapshot_since(cursor) { - return (delta, pos, false); + // A retained delta within budget is byte-exact and appends + // (reset:false). An oversized delta is capped to a bounded tail + // and reset:true — the client's xterm cannot keep more than its + // scrollback anyway, so an over-budget delta would only waste + // parse above an already-stale screen. + match tail_bytes { + Some(limit) if delta.len() > limit => { + return (sb.snapshot_tail(limit), pos, true); + } + _ => return (delta, pos, false), + } } - // Cursor aged out of the ring: a full reset snapshot, untrimmed — - // the tail cap is a cold-attach affordance and a warm reattach is - // never quietly narrowed. - return (sb.snapshot(), pos, true); + // Cursor aged out of the ring: a bounded, ground-state-aligned tail + // and a reset — never the whole untrimmed ring. + let snapshot = match tail_bytes { + Some(limit) => sb.snapshot_tail(limit), + None => sb.snapshot(), + }; + return (snapshot, pos, true); } // Cold attach: bound the snapshot to the client's tail hint so first // open never ships the whole ring buffer. diff --git a/engine/server/src/ws.rs b/engine/server/src/ws.rs index 86c1f49a..89b16f4a 100644 --- a/engine/server/src/ws.rs +++ b/engine/server/src/ws.rs @@ -99,6 +99,47 @@ fn coalesce(snap_pos: u64, chunks: &[OutputChunk]) -> (Vec, u64) { (out, end) } +/// Send whatever broadcast chunks are already queued for this socket, without +/// blocking, advancing `sent_pos` past them. Used before answering a liveness +/// probe so the pong's position reflects output the client has actually been +/// sent. A `Lagged`/`Closed` on the non-blocking drain is left for the main +/// `rx.recv()` loop to handle (resync or teardown); flushing what was already +/// pulled is still correct and gap-free. +async fn flush_available( + sink: &mut S, + rx: &mut tokio::sync::broadcast::Receiver, + snap_pos: u64, + sent_pos: &mut u64, +) -> Result<(), ()> +where + S: SinkExt + Unpin, +{ + let mut drained: Vec = Vec::new(); + let mut queued = 0usize; + while queued < OUTBOUND_COALESCE_CAP { + match rx.try_recv() { + Ok(next) => { + queued += next.data.len(); + drained.push(next); + } + Err(_) => break, + } + } + if drained.is_empty() { + return Ok(()); + } + let (frame, end) = coalesce(snap_pos, &drained); + if end > *sent_pos { + *sent_pos = end; + } + if !frame.is_empty() { + sink.send(Message::Binary(frame.into())) + .await + .map_err(|_| ())?; + } + Ok(()) +} + /// Re-synchronise a lagging client in-band, on the same socket, instead of /// dropping it and forcing a reconnect + replay (a cascade under load). /// @@ -315,7 +356,11 @@ async fn handle_socket( }; let (mut sink, mut stream) = socket.split(); - let (control_tx, mut control_rx) = mpsc::unbounded_channel::(); + // Carries liveness-probe ids from the inbound task to the outbound task. + // The pong is *formed* by the outbound task with the position it has + // actually streamed, not by the inbound task with `total_written` — see the + // ping branch of the outbound select (WI-126). + let (ping_tx, mut ping_rx) = mpsc::unbounded_channel::(); // Subscribe BEFORE snapshotting so no broadcast chunks are missed in the gap. let mut rx = session.subscribe(); @@ -386,10 +431,11 @@ async fn handle_socket( let _ = writer_session.resize(cols, rows); } Ok(ClientControl::Ping { id }) => { - let _ = control_tx.send(ServerControl::Pong { - id, - pos: writer_session.scrollback_position(), - }); + // Hand the probe to the outbound task; it answers + // with the position it has actually sent this + // socket, after flushing queued output, so the pong + // can never report the server ahead of the client. + let _ = ping_tx.send(id); } Ok(ClientControl::Auth { .. }) => { // Already authenticated; ignore further auth frames. @@ -427,9 +473,23 @@ async fn handle_socket( loop { tokio::select! { - Some(control) = control_rx.recv() => { + Some(ping_id) = ping_rx.recv() => { + // Flush anything already queued for this socket, then answer + // the probe with `sent_pos` — the byte offset actually + // written here. Because the pong is ordered AFTER those + // chunks on the wire and carries only what has been sent, it + // can never report the server ahead of what the client has + // received, so a bursty session no longer trips a spurious + // recycle (WI-126). + if flush_available(&mut sink, &mut rx, snap_pos, &mut sent_pos) + .await + .is_err() + { + break; + } + let pong = ServerControl::Pong { id: ping_id, pos: sent_pos }; if sink - .send(Message::Text(serde_json::to_string(&control).unwrap().into())) + .send(Message::Text(serde_json::to_string(&pong).unwrap().into())) .await .is_err() { diff --git a/engine/server/tests/integration.rs b/engine/server/tests/integration.rs index 539fa114..535e4507 100644 --- a/engine/server/tests/integration.rs +++ b/engine/server/tests/integration.rs @@ -82,6 +82,21 @@ async fn boot() -> (String, tokio::task::JoinHandle<()>) { boot_with_config(test_config()).await } +/// Boot with a chosen scrollback ring size, leaving every other test-config +/// default in place. The suite default stays 64 KiB (`test_config`); the +/// large-session resume tests use this to size the ring around the flood they +/// are exercising (a tiny ring an early cursor ages out of, or the 4 MiB ring +/// prod runs). A per-session `scrollback_bytes` override on session-create also +/// works, but sizing the whole engine keeps each test's intent in one place. +#[allow(dead_code)] // used by the large-session resume tests +async fn boot_with(scrollback_bytes: usize) -> (String, tokio::task::JoinHandle<()>) { + boot_with_config(Config { + scrollback_bytes, + ..test_config() + }) + .await +} + async fn boot_with_config(cfg: Config) -> (String, tokio::task::JoinHandle<()>) { let (base, _state, handle) = boot_with_state(cfg).await; (base, handle) @@ -3128,18 +3143,29 @@ async fn ws_attach_with_auth( } /// Read one snapshot sequence (`snapshot-start` → binary frames → -/// `snapshot-done`) and report whether it was a reset and how many snapshot -/// bytes were streamed. +/// `snapshot-done`) and report whether it was a reset, how many snapshot bytes +/// were streamed, and the leading bytes of the payload. +/// +/// The leading bytes let a caller assert the ground-state alignment invariant +/// the ring guarantees: a snapshot never begins on a UTF-8 continuation byte, +/// and never in the tail of a chopped escape sequence. Up to +/// [`SNAPSHOT_HEAD_BYTES`] are captured — enough to recognise a `\x1b[` CSI +/// introducer at the start (a *complete* sequence, not a fragment) versus a +/// bare `[`/`m`/digit left over from a mid-sequence cut. +const SNAPSHOT_HEAD_BYTES: usize = 64; + +#[allow(dead_code)] // `first_bytes` is only read by the large-session tests. async fn read_snapshot( ws: &mut tokio_tungstenite::WebSocketStream< tokio_tungstenite::MaybeTlsStream, >, -) -> (bool, usize) { +) -> (bool, usize, Vec) { let mut reset = false; let mut started = false; let mut total = 0usize; + let mut first_bytes: Vec = Vec::new(); loop { - let m = tokio::time::timeout(Duration::from_secs(3), ws.next()) + let m = tokio::time::timeout(Duration::from_secs(5), ws.next()) .await .expect("snapshot frame arrives") .unwrap() @@ -3156,11 +3182,40 @@ async fn read_snapshot( _ => {} } } - Message::Binary(b) if started => total += b.len(), + Message::Binary(b) if started => { + if first_bytes.len() < SNAPSHOT_HEAD_BYTES { + let take = (SNAPSHOT_HEAD_BYTES - first_bytes.len()).min(b.len()); + first_bytes.extend_from_slice(&b[..take]); + } + total += b.len(); + } _ => {} } } - (reset, total) + (reset, total, first_bytes) +} + +/// A snapshot begins in the terminal's ground state: its first byte is never a +/// UTF-8 continuation byte, and if it opens an escape it is a whole CSI/OSC +/// introducer (`\x1b` followed by `[` or `]`), never the tail of one chopped by +/// an overflow cut (which would begin with a bare `[`, a digit, or `m`). +fn assert_ground_state_aligned(first_bytes: &[u8]) { + if first_bytes.is_empty() { + return; + } + let b0 = first_bytes[0]; + assert!( + (b0 & 0xC0) != 0x80, + "snapshot must not begin on a UTF-8 continuation byte; got {b0:#04x}" + ); + if b0 == 0x1b { + assert!( + matches!(first_bytes.get(1), Some(b'[') | Some(b']')), + "a leading ESC must introduce a complete CSI/OSC sequence, not a \ + fragment; got {:?}", + &first_bytes[..first_bytes.len().min(4)] + ); + } } /// Drive more than `at_least` bytes of output into a `cat` session's ring, as @@ -3240,7 +3295,7 @@ async fn cold_attach_tail_hint_caps_the_snapshot() { json!({ "type": "auth", "token": TEST_TOKEN, "snapshot_tail_bytes": TAIL }), ) .await; - let (reset, len) = read_snapshot(&mut cold).await; + let (reset, len, _head) = read_snapshot(&mut cold).await; assert!(reset, "a cold attach is always a full reset"); assert!( len > 0 && len <= TAIL, @@ -3251,10 +3306,12 @@ async fn cold_attach_tail_hint_caps_the_snapshot() { } #[tokio::test] -async fn warm_attach_is_not_narrowed_by_a_tail_hint() { - // The tail cap is a cold-attach affordance. A warm reattach carries - // `resume_from`; even if a tail hint is also present it must be ignored and - // the resume delta returned in full (`reset: false`). +async fn warm_attach_over_budget_is_a_bounded_reset() { + // F1 (WI-125) inverts the old `warm_attach_is_not_narrowed_by_a_tail_hint`. + // The tail hint is now the replay budget on *every* path. A warm reattach + // whose delta exceeds the budget no longer floods the whole ring: it is + // capped to a ground-state tail and reset:true, because the client's xterm + // cannot keep more than its scrollback anyway. let (base, _h) = boot().await; let client = reqwest::Client::builder() .default_headers(auth()) @@ -3277,8 +3334,8 @@ async fn warm_attach_is_not_narrowed_by_a_tail_hint() { const TAIL: usize = 512; fill_scrollback(&base, &id, 4 * TAIL).await; - // resume_from = 0 resolves to the whole retained ring as a delta. With a - // tail hint also set, the warm path must still return it untrimmed. + // resume_from = 0 resolves to the whole retained ring as a delta — far more + // than the 512-byte budget — so F1 caps it to a bounded reset. let mut warm = ws_attach_with_auth( &base, &id, @@ -3290,16 +3347,624 @@ async fn warm_attach_is_not_narrowed_by_a_tail_hint() { }), ) .await; - let (reset, len) = read_snapshot(&mut warm).await; - assert!(!reset, "a warm resume from 0 is a delta, not a reset"); + let (reset, len, head) = read_snapshot(&mut warm).await; assert!( - len > TAIL, - "warm reattach must ignore the tail hint; got {len} <= {TAIL}" + reset, + "an over-budget delta is a bounded reset, not an append" ); + assert!( + len > 0 && len <= TAIL, + "the over-budget delta must be capped to the tail budget ({TAIL}); got {len}" + ); + assert_ground_state_aligned(&head); warm.close(None).await.ok(); kill_session(&client, &base, &id).await; } +#[tokio::test] +async fn warm_attach_within_budget_stays_a_byte_exact_delta() { + // The companion to the bounded-reset case: a warm reattach whose delta fits + // the budget is still sent byte-for-byte with reset:false, so an ordinary + // switch-away/switch-back appends without clearing the terminal. + let (base, _h) = boot().await; + let client = reqwest::Client::builder() + .default_headers(auth()) + .build() + .unwrap(); + + let id: String = client + .post(format!("{base}/api/sessions")) + .json(&json!({ + "name": "warm-small", + "command": ["/bin/sh", "-c", "stty -echo; exec /bin/cat"] + })) + .send() + .await + .unwrap() + .json::() + .await + .unwrap()["id"] + .as_str() + .unwrap() + .to_string(); + + // Prime the ring, record the cursor, then produce a small delta. + let mut ws = ws_attach(&base, &id).await; + read_snapshot(&mut ws).await; + ws.send(Message::Binary(b"before-cursor\n".to_vec().into())) + .await + .unwrap(); + let _ = collect_binary_until(&mut ws, b"before-cursor", Duration::from_secs(2)).await; + let cursor = { + let detail: Value = client + .get(format!("{base}/api/sessions/{id}")) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + detail["scrollback_pos"].as_u64().unwrap() + }; + ws.send(Message::Binary(b"after-cursor\n".to_vec().into())) + .await + .unwrap(); + let _ = collect_binary_until(&mut ws, b"after-cursor", Duration::from_secs(2)).await; + ws.close(None).await.ok(); + + // A generous budget the tiny delta fits inside: the resume stays a delta. + let mut warm = ws_attach_with_auth( + &base, + &id, + json!({ + "type": "auth", + "token": TEST_TOKEN, + "resume_from": cursor, + "snapshot_tail_bytes": 1024 * 1024, + }), + ) + .await; + let (reset, delta) = read_snapshot_payload(&mut warm).await; + assert!(!reset, "a delta within budget must not reset the terminal"); + assert!( + delta + .windows(b"after-cursor".len()) + .any(|w| w == b"after-cursor"), + "the delta must carry the post-cursor output" + ); + assert!( + !delta + .windows(b"before-cursor".len()) + .any(|w| w == b"before-cursor"), + "the delta must exclude pre-cursor output (byte-exact from the cursor)" + ); + warm.close(None).await.ok(); + kill_session(&client, &base, &id).await; +} + +// ---- large-session resume harness (WI-122 / H1) ---- +// +// These tests drive real bytes through the PTY reader, the 1024-slot broadcast +// channel and the outbound coalesce path via a load-generating session, then +// exercise resume across a ring the cursor has aged out of, and the in-band +// lag recovery. Two of them pin *today's* flood behaviour with a comment +// pointing at the F1 (WI-125) inversion; the ground-state-alignment, +// byte-exact-delta and no-duplicate assertions hold across that change. + +/// A load-generating session command: a bash loop emitting `count` coloured, +/// sequence-numbered lines — a monotonic `SEQNNNNNNNN` spine so a replay can be +/// checked for gaps and duplicates — with a periodic cursor-home frame so the +/// stream carries real CSI sequences, not just plain text. After the burst it +/// idles (`sleep`) so the ring is stable while a test reattaches; every caller +/// reaps it with `kill_session`. Each line is ~65 bytes. +fn load_session_command(count: u32) -> Vec { + vec![ + "/bin/bash".into(), + "-c".into(), + format!( + "i=0; while [ \"$i\" -lt {count} ]; do \ + printf '\\033[3%dmSEQ%08d the quick brown fox jumps over the lazy dog\\033[0m\\n' \ + \"$((i % 8))\" \"$i\"; \ + i=$((i + 1)); \ + if [ \"$((i % 64))\" -eq 0 ]; then printf '\\033[H'; fi; \ + done; \ + exec sleep 3600" + ), + ] +} + +/// Create a session running `command`, returning its id. +async fn create_command_session( + client: &reqwest::Client, + base: &str, + name: &str, + command: Vec, +) -> String { + client + .post(format!("{base}/api/sessions")) + .json(&json!({ "name": name, "command": command })) + .send() + .await + .unwrap() + .json::() + .await + .unwrap()["id"] + .as_str() + .unwrap() + .to_string() +} + +/// Poll `GET /api/sessions/:id` until `scrollback_pos` reaches `target`, +/// returning the observed position. Fails on a deadline so a session that never +/// produces enough fails rather than hanging. +async fn poll_scrollback_at_least( + client: &reqwest::Client, + base: &str, + id: &str, + target: u64, +) -> u64 { + let deadline = tokio::time::Instant::now() + Duration::from_secs(8); + loop { + let detail: SessionDetail = client + .get(format!("{base}/api/sessions/{id}")) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + let pos = detail.scrollback_pos; + if pos >= target { + return pos; + } + assert!( + tokio::time::Instant::now() < deadline, + "session {id} produced only {pos} bytes, wanted >= {target}" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +/// Wait until `scrollback_pos` stops advancing (two equal reads 80 ms apart), +/// so a test that compares two reattaches resolves both cursors against the +/// same, stable ring. Returns the settled position. +async fn wait_until_scrollback_stable(client: &reqwest::Client, base: &str, id: &str) -> u64 { + let deadline = tokio::time::Instant::now() + Duration::from_secs(8); + let mut last = u64::MAX; + loop { + let detail: SessionDetail = client + .get(format!("{base}/api/sessions/{id}")) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + let pos = detail.scrollback_pos; + if pos == last { + return pos; + } + last = pos; + assert!( + tokio::time::Instant::now() < deadline, + "session {id} never stopped producing (last {pos})" + ); + tokio::time::sleep(Duration::from_millis(80)).await; + } +} + +/// Read one whole snapshot sequence, returning `(reset, full_payload)`. Unlike +/// [`read_snapshot`] this keeps every byte, so a caller can compare two deltas +/// for byte-exactness. +async fn read_snapshot_payload( + ws: &mut tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, + >, +) -> (bool, Vec) { + let mut reset = false; + let mut started = false; + let mut payload = Vec::new(); + loop { + let m = tokio::time::timeout(Duration::from_secs(5), ws.next()) + .await + .expect("snapshot frame arrives") + .unwrap() + .unwrap(); + match m { + Message::Text(s) => { + let v: Value = serde_json::from_str(&s).unwrap(); + match v["type"].as_str() { + Some("snapshot-start") => { + started = true; + reset = v["reset"].as_bool().unwrap_or(true); + } + Some("snapshot-done") => break, + _ => {} + } + } + Message::Binary(b) if started => payload.extend_from_slice(&b), + _ => {} + } + } + (reset, payload) +} + +/// One delivered snapshot/resync sequence: whether it reset the terminal and +/// the payload bytes that followed it. +struct SnapSegment { + reset: bool, + bytes: Vec, +} + +/// Drain frames, grouping the payload after each `snapshot-start` into its own +/// segment, until no frame arrives for `idle`. Returns the initial snapshot, +/// the live bytes that followed it, and any in-band resync the server sent — +/// each `snapshot-start` opens a new segment, so a resync is a fresh segment. +async fn read_segments_until_idle( + ws: &mut tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, + >, + idle: Duration, +) -> Vec { + let mut segs: Vec = Vec::new(); + loop { + match tokio::time::timeout(idle, ws.next()).await { + Err(_) | Ok(None) | Ok(Some(Err(_))) => break, + Ok(Some(Ok(m))) => match m { + Message::Text(s) => { + let v: Value = serde_json::from_str(&s).unwrap(); + if v["type"] == "snapshot-start" { + segs.push(SnapSegment { + reset: v["reset"].as_bool().unwrap_or(true), + bytes: Vec::new(), + }); + } + } + Message::Binary(b) => { + if let Some(seg) = segs.last_mut() { + seg.bytes.extend_from_slice(&b); + } + } + _ => {} + }, + } + } + segs +} + +/// Extract, in order, every complete `SEQ%08d` marker in `bytes`. The producer +/// emits each sequence number exactly once and monotonically, so this is the +/// spine used to detect gaps and duplicates in a replay. +fn seq_numbers(bytes: &[u8]) -> Vec { + let mut out = Vec::new(); + let mut i = 0; + while i + 11 <= bytes.len() { + if &bytes[i..i + 3] == b"SEQ" && bytes[i + 3..i + 11].iter().all(u8::is_ascii_digit) { + let n: u32 = std::str::from_utf8(&bytes[i + 3..i + 11]) + .unwrap() + .parse() + .unwrap(); + out.push(n); + i += 11; + } else { + i += 1; + } + } + out +} + +/// Assert a run of `SEQ` markers is contiguous: strictly increasing by exactly +/// one, no gap and no duplicate. Used on a delta that resumed from a retained +/// cursor, where every intervening line must be present exactly once. +fn assert_seq_spine_contiguous(bytes: &[u8]) { + let seqs = seq_numbers(bytes); + assert!( + seqs.len() > 10, + "expected a run of SEQ markers in the delta, saw {}", + seqs.len() + ); + for w in seqs.windows(2) { + assert_eq!( + w[1], + w[0] + 1, + "SEQ spine is not contiguous ({} then {}): a gap or duplicate in the delta", + w[0], + w[1] + ); + } +} + +#[tokio::test] +async fn large_stale_warm_reattach_is_a_bounded_reset() { + // H1(a) / F1. Ring 64 KiB. A load session pushes far past capacity, so a + // cursor recorded early has aged out of the ring by the time we reattach + // with it. Before F1 `snapshot_for_attach` answered that stale warm + // reattach with the FULL untrimmed ring (the flood, 4x worse than a cold + // attach). F1 (WI-125) bounds it: a ground-state-aligned tail of at most + // the budget, still reset:true. + const CAP: usize = 64 * 1024; + const TAIL_BUDGET: usize = 8 * 1024; // the F1 budget, well under CAP + let (base, _h) = boot_with(CAP).await; + let client = reqwest::Client::builder() + .default_headers(auth()) + .build() + .unwrap(); + + let id = create_command_session(&client, &base, "flood", load_session_command(20_000)).await; + + // Record an early cursor, then wait until the ring has advanced several + // capacities past it so it is unambiguously aged out. + let early = poll_scrollback_at_least(&client, &base, &id, CAP as u64 / 2).await; + let _ = poll_scrollback_at_least(&client, &base, &id, early + 4 * CAP as u64).await; + + let mut warm = ws_attach_with_auth( + &base, + &id, + json!({ + "type": "auth", + "token": TEST_TOKEN, + "resume_from": early, + "snapshot_tail_bytes": TAIL_BUDGET, + }), + ) + .await; + let (reset, len, head) = read_snapshot(&mut warm).await; + assert!(reset, "a cursor aged out of the ring forces a full reset"); + assert!( + len > 0 && len <= TAIL_BUDGET, + "F1 bounds the aged-out reattach to the tail budget ({TAIL_BUDGET}); got {len}" + ); + assert_ground_state_aligned(&head); + warm.close(None).await.ok(); + kill_session(&client, &base, &id).await; +} + +#[tokio::test] +async fn four_mib_ring_stale_reattach_is_ground_state_aligned_and_bounded() { + // H1(b) / F1. The dimensions prod runs: a 4 MiB ring with ~6.5 MiB pushed. + // The stale warm reattach is bounded to the tail budget and reset:true, + // aligned to the terminal's ground state — not the whole 4 MiB ring. + const CAP: usize = 4 * 1024 * 1024; + const TAIL_BUDGET: usize = 1024 * 1024; + let (base, _h) = boot_with(CAP).await; + let client = reqwest::Client::builder() + .default_headers(auth()) + .build() + .unwrap(); + + let id = + create_command_session(&client, &base, "flood-4m", load_session_command(100_000)).await; + + // Push well past the ring, then reattach from position 1 (long aged out). + let _ = poll_scrollback_at_least(&client, &base, &id, CAP as u64 + CAP as u64 / 2).await; + let mut warm = ws_attach_with_auth( + &base, + &id, + json!({ + "type": "auth", + "token": TEST_TOKEN, + "resume_from": 1, + "snapshot_tail_bytes": TAIL_BUDGET, + }), + ) + .await; + let (reset, len, head) = read_snapshot(&mut warm).await; + assert!(reset, "position 1 has long since aged out of a 4 MiB ring"); + assert!( + len > 0 && len <= TAIL_BUDGET, + "F1 bounds the aged-out 4 MiB reattach to the tail budget ({TAIL_BUDGET}); got {len}" + ); + assert_ground_state_aligned(&head); + warm.close(None).await.ok(); + kill_session(&client, &base, &id).await; +} + +#[tokio::test] +async fn retained_cursor_deltas_are_byte_exact_under_load() { + // H1(b) — byte-exactness half. Two warm reattaches from retained cursors + // Ca < Cb into the same idle 4 MiB ring: the later delta must equal the + // earlier delta with its first (Cb-Ca) bytes removed, and its SEQ spine + // must be contiguous. Proves `snapshot_since` is byte-exact against bytes + // that really flowed through the PTY reader and broadcast path, with no gap + // or duplicate at the resume boundary. This assertion survives F1. + const CAP: usize = 4 * 1024 * 1024; + let (base, _h) = boot_with(CAP).await; + let client = reqwest::Client::builder() + .default_headers(auth()) + .build() + .unwrap(); + + let id = create_command_session(&client, &base, "delta", load_session_command(80_000)).await; + // Let it settle so both cursors resolve against the same stable ring. + let _ = poll_scrollback_at_least(&client, &base, &id, CAP as u64 + 256 * 1024).await; + let total = wait_until_scrollback_stable(&client, &base, &id).await; + + // Both cursors sit inside the retained 4 MiB window. + let ca = total - 2 * 1024 * 1024; + let cb = ca + 512 * 1024; + + let mut a = ws_attach_with_token_and_cursor(&base, &id, TEST_TOKEN, Some(ca)).await; + let (reset_a, delta_a) = read_snapshot_payload(&mut a).await; + let mut b = ws_attach_with_token_and_cursor(&base, &id, TEST_TOKEN, Some(cb)).await; + let (reset_b, delta_b) = read_snapshot_payload(&mut b).await; + + assert!(!reset_a && !reset_b, "retained cursors resume as deltas"); + let skip = (cb - ca) as usize; + assert!( + delta_a.len() >= skip && delta_a.len() == skip + delta_b.len(), + "delta lengths inconsistent: |A|={} skip={} |B|={}", + delta_a.len(), + skip, + delta_b.len() + ); + assert_eq!( + &delta_a[skip..], + &delta_b[..], + "the later delta must equal the earlier delta past the (Cb-Ca) boundary" + ); + assert_seq_spine_contiguous(&delta_b); + + a.close(None).await.ok(); + b.close(None).await.ok(); + kill_session(&client, &base, &id).await; +} + +#[tokio::test] +async fn lagging_subscriber_recovers_in_band_bounded_and_without_duplicates() { + // H1(c) — lag recovery. A client stops reading while a load session floods + // far past the 1024-slot broadcast channel, so the server must recover the + // socket in-band (`send_resync`) rather than drop it. The ring is sized + // above the ~8 MiB broadcast overflow so the resync stays a reset:false + // delta. Assert: a resync segment appears, every segment's payload is + // bounded by the ring, and the SEQ spine across the whole delivered stream + // is strictly increasing — no duplicate and no reordering across the + // resync boundary. + const CAP: usize = 12 * 1024 * 1024; + let (base, _h) = boot_with(CAP).await; + let client = reqwest::Client::builder() + .default_headers(auth()) + .build() + .unwrap(); + + // ~11.7 MiB of output, comfortably past the broadcast channel capacity. + let id = create_command_session(&client, &base, "lag", load_session_command(180_000)).await; + + let mut ws = ws_attach(&base, &id).await; + // Do not read at all: the initial snapshot and the live stream both queue + // in the socket buffer, the outbound task blocks on a full socket, and the + // broadcast channel overflows behind it. + tokio::time::sleep(Duration::from_millis(1500)).await; + // Resume and drain everything, including the in-band resync. + let segs = read_segments_until_idle(&mut ws, Duration::from_millis(1500)).await; + + assert!( + segs.len() >= 2, + "expected an in-band resync segment after the flood; saw {} segment(s)", + segs.len() + ); + assert!( + segs[0].reset, + "the initial cold attach is always a full reset snapshot" + ); + for s in &segs { + assert!( + s.bytes.len() <= CAP + CAP / 16, + "a resync/snapshot payload must be bounded by the ring; got {}", + s.bytes.len() + ); + } + // The resync snapshots from the client's last sent position, so it never + // re-sends bytes already delivered: the spine is strictly increasing. + let mut all = Vec::new(); + for s in &segs { + all.extend(seq_numbers(&s.bytes)); + } + assert!( + all.len() > 100, + "expected a long SEQ spine across the delivered stream, saw {}", + all.len() + ); + for w in all.windows(2) { + assert!( + w[1] > w[0], + "SEQ spine went backward or duplicated across a resync: {} then {}", + w[0], + w[1] + ); + } + + ws.close(None).await.ok(); + kill_session(&client, &base, &id).await; +} + +#[tokio::test] +async fn pong_never_reports_the_server_ahead_of_this_socket() { + // F2 (WI-126). Under load the liveness pong must carry the position actually + // streamed to THIS socket, never `total_written` (which includes queued but + // unsent output). Before the fix the inbound task answered with + // scrollback_position() and the pong could be delivered ahead of the chunks + // it referenced, so the client saw the server "ahead" and recycled the + // socket. Now the outbound task answers with sent_pos after flushing, so the + // pong's pos is never greater than what this socket has received. + let (base, _h) = boot_with(4 * 1024 * 1024).await; + let client = reqwest::Client::builder() + .default_headers(auth()) + .build() + .unwrap(); + let id = create_command_session(&client, &base, "pong", load_session_command(50_000)).await; + + let mut ws = ws_attach(&base, &id).await; + + // Drain the initial snapshot, recording the absolute position it ended at. + let mut snap_end: u64 = 0; + let mut in_snapshot = false; + loop { + let m = tokio::time::timeout(Duration::from_secs(5), ws.next()) + .await + .expect("snapshot frame arrives") + .unwrap() + .unwrap(); + match m { + Message::Text(s) => { + let v: Value = serde_json::from_str(&s).unwrap(); + match v["type"].as_str() { + Some("snapshot-start") => { + in_snapshot = true; + snap_end = v["scrollback_pos"].as_u64().unwrap(); + } + Some("snapshot-done") => break, + _ => {} + } + } + Message::Binary(_) if in_snapshot => {} + _ => {} + } + } + + // Probe while output is still flowing. + ws.send(Message::Text( + json!({ "type": "ping", "id": 1 }).to_string().into(), + )) + .await + .unwrap(); + + // Count live bytes received on this socket until the pong arrives. The pong + // is ordered after any flushed chunks, so by the time we read it we have + // received every byte up to its position. + let mut live: u64 = 0; + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + let m = tokio::time::timeout(remaining, ws.next()) + .await + .expect("a pong should arrive within the deadline") + .unwrap() + .unwrap(); + match m { + Message::Binary(b) => live += b.len() as u64, + Message::Text(s) => { + let v: Value = serde_json::from_str(&s).unwrap(); + if v["type"] == "pong" { + assert_eq!(v["id"].as_u64(), Some(1)); + let pos = v["pos"].as_u64().unwrap(); + let received = snap_end + live; + assert!( + pos <= received, + "pong pos {pos} is ahead of what this socket received \ + ({received} = snap_end {snap_end} + live {live})" + ); + break; + } + // A resync snapshot-start would only add to `received`; keep going. + } + _ => {} + } + } + + ws.close(None).await.ok(); + kill_session(&client, &base, &id).await; +} + #[tokio::test] async fn ws_attach_echoes_input_and_replays_on_reattach() { let (base, _h) = boot().await; diff --git a/scripts/capture_transcript.py b/scripts/capture_transcript.py new file mode 100755 index 00000000..24eff752 --- /dev/null +++ b/scripts/capture_transcript.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +"""Capture the raw PTY output of a command to a file. + +The fidelity fixtures under web/tests/fixtures/transcripts/ are ideally captured +from a real dev-stack session via `GET /api/history/:id/download`. When no stack +is reachable (an agent session cannot attach to prod), this runs a real program +under a PTY on this box instead, so the bytes carry genuine escape sequences, +SGR colour, cursor moves, alt-screen switches and UTF-8 — exactly what the +parser-fidelity tests exercise — rather than anything synthetic. + +Run the result through scripts/sanitise_transcript.py before committing it. + +Usage: + capture_transcript.py OUTPUT.bin [--max-bytes N] -- CMD [ARG ...] +""" + +from __future__ import annotations + +import argparse +import contextlib +import os +import pty +import select +import sys +from pathlib import Path + + +def capture(cmd: list[str], max_bytes: int, cols: int = 120, rows: int = 40) -> bytes: + pid, fd = pty.fork() + if pid == 0: # child + os.environ["TERM"] = "xterm-256color" + os.environ["COLUMNS"] = str(cols) + os.environ["LINES"] = str(rows) + try: + os.execvp(cmd[0], cmd) + except FileNotFoundError: + os._exit(127) + # parent: set window size then drain until EOF or the byte cap. + try: + import fcntl + import struct + import termios + + fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0)) + except Exception: + pass + + chunks: list[bytes] = [] + total = 0 + while total < max_bytes: + try: + readable, _, _ = select.select([fd], [], [], 10.0) + except OSError: + break + if not readable: + break + try: + data = os.read(fd, 65536) + except OSError: + break + if not data: + break + chunks.append(data) + total += len(data) + with contextlib.suppress(OSError): + os.close(fd) + with contextlib.suppress(OSError): + os.waitpid(pid, 0) + return b"".join(chunks)[:max_bytes] + + +def main(argv: list[str]) -> int: + # Split on the first "--" ourselves: argparse.REMAINDER would greedily + # swallow --max-bytes into the command. + if "--" in argv: + sep = argv.index("--") + pre, cmd = argv[:sep], argv[sep + 1 :] + else: + pre, cmd = argv, [] + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("output") + ap.add_argument("--max-bytes", type=int, default=2 * 1024 * 1024) + args = ap.parse_args(pre) + if not cmd: + ap.error("a command is required after --") + data = capture(cmd, args.max_bytes) + Path(args.output).write_bytes(data) + print(f"capture_transcript: {len(data)} bytes -> {args.output}") + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/scripts/sanitise_transcript.py b/scripts/sanitise_transcript.py new file mode 100755 index 00000000..b3dddb98 --- /dev/null +++ b/scripts/sanitise_transcript.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""Sanitise a captured terminal transcript before it is checked in as a fixture. + +A transcript is raw PTY bytes -- escape sequences, colour, UTF-8, whatever a +program wrote. Captured from a real session it can carry things a fixture must +not: absolute home paths, a bearer token echoed onto a command line, an email +address, an API key in an environment dump. This pass rewrites those to inert +placeholders in place in the byte stream and then *asserts* that nothing +matching a secret pattern survives, so a fixture can never be committed with a +live credential in it. + +Usage: + sanitise_transcript.py INPUT.bin OUTPUT.bin + sanitise_transcript.py --check INPUT.bin # assert only, no write + +Exit non-zero if a secret pattern remains after rewriting. +""" + +from __future__ import annotations + +import argparse +import re +import sys +from pathlib import Path + +# (pattern, replacement) applied in order, over the raw bytes. Replacements are +# kept close to the original length so escape-sequence and line boundaries stay +# meaningful for the fidelity tests. +_REWRITES: list[tuple[re.Pattern[bytes], bytes]] = [ + # Bearer / auth tokens on a command line or in a header. + ( + re.compile(rb"(?i)(authorization:\s*bearer\s+)[A-Za-z0-9._\-]+"), + rb"\1REDACTED_TOKEN", + ), + ( + re.compile(rb"(?i)(token[=:\s\"']+)[A-Za-z0-9._\-]{16,}"), + rb"\1REDACTED_TOKEN", + ), + # Generic long secrets: sk-..., ghp_..., AKIA..., xoxb-..., JWTs. + (re.compile(rb"sk-[A-Za-z0-9]{20,}"), rb"sk-REDACTED"), + (re.compile(rb"gh[pousr]_[A-Za-z0-9]{20,}"), rb"ghx_REDACTED"), + (re.compile(rb"AKIA[0-9A-Z]{16}"), rb"AKIAREDACTEDREDACT00"), + (re.compile(rb"xox[baprs]-[A-Za-z0-9-]{10,}"), rb"xoxb-REDACTED"), + ( + re.compile( + rb"eyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}" + ), + rb"eyJ.REDACTED.JWT", + ), + # Email addresses. + ( + re.compile(rb"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}"), + rb"user@example.invalid", + ), + # Absolute home paths -> a neutral placeholder. + (re.compile(rb"/home/[A-Za-z0-9._\-]+"), rb"/home/user"), + (re.compile(rb"/Users/[A-Za-z0-9._\-]+"), rb"/Users/user"), +] + +# After rewriting, NONE of these may appear. If one does, the sanitiser failed +# to cover a case and the fixture is refused rather than committed with a leak. +_FORBIDDEN: list[re.Pattern[bytes]] = [ + re.compile(rb"(?i)bearer\s+[A-Za-z0-9._\-]{16,}"), + re.compile(rb"sk-[A-Za-z0-9]{20,}"), + re.compile(rb"gh[pousr]_[A-Za-z0-9]{20,}"), + re.compile(rb"AKIA[0-9A-Z]{16}"), + re.compile(rb"xox[baprs]-[A-Za-z0-9-]{10,}"), + re.compile(rb"eyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}"), + # A home path for a real user (the placeholder "user" is allowed). + re.compile(rb"/home/(?!user\b)[A-Za-z0-9._\-]+"), + re.compile(rb"/Users/(?!user\b)[A-Za-z0-9._\-]+"), +] + + +def sanitise(data: bytes) -> bytes: + for pattern, repl in _REWRITES: + data = pattern.sub(repl, data) + return data + + +def assert_clean(data: bytes) -> None: + leaks = [] + for pattern in _FORBIDDEN: + match = pattern.search(data) + if match: + leaks.append(f"{pattern.pattern!r} matched {match.group(0)[:32]!r}") + if leaks: + raise SystemExit( + "sanitise_transcript: secret pattern survived sanitisation:\n " + + "\n ".join(leaks) + ) + + +def main(argv: list[str]) -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("input") + ap.add_argument("output", nargs="?") + ap.add_argument("--check", action="store_true", help="assert only; do not write") + args = ap.parse_args(argv) + + data = Path(args.input).read_bytes() + + if args.check: + assert_clean(data) + print(f"sanitise_transcript: {args.input} clean ({len(data)} bytes)") + return 0 + + if not args.output: + ap.error("OUTPUT is required unless --check is given") + cleaned = sanitise(data) + assert_clean(cleaned) + out = Path(args.output) + out.resolve().parent.mkdir(parents=True, exist_ok=True) + out.write_bytes(cleaned) + print( + f"sanitise_transcript: {args.input} -> {args.output} " + f"({len(data)} -> {len(cleaned)} bytes, clean)" + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/web/package.json b/web/package.json index c50329c1..6cdc30fc 100644 --- a/web/package.json +++ b/web/package.json @@ -22,6 +22,7 @@ "@solidjs/router": "^0.16.1", "@xterm/addon-fit": "^0.11.0", "@xterm/addon-search": "^0.16.0", + "@xterm/addon-serialize": "^0.14.0", "@xterm/addon-web-links": "^0.12.0", "@xterm/xterm": "^6.0.0", "monaco-editor": "^0.55.1", @@ -32,6 +33,8 @@ "@solidjs/testing-library": "^0.8.10", "@testing-library/jest-dom": "^7.0.1", "@testing-library/user-event": "^14.6.4", + "@xterm/headless": "^6.0.0", + "fake-indexeddb": "^6.2.5", "jsdom": "^30.0.1", "typescript": "^6.0.3", "vite": "^8.2.2", diff --git a/web/pnpm-lock.yaml b/web/pnpm-lock.yaml index 08d3ad28..ff9e1c55 100644 --- a/web/pnpm-lock.yaml +++ b/web/pnpm-lock.yaml @@ -34,6 +34,9 @@ importers: '@xterm/addon-search': specifier: ^0.16.0 version: 0.16.0 + '@xterm/addon-serialize': + specifier: ^0.14.0 + version: 0.14.0 '@xterm/addon-web-links': specifier: ^0.12.0 version: 0.12.0 @@ -59,6 +62,12 @@ importers: '@testing-library/user-event': specifier: ^14.6.4 version: 14.6.4(@testing-library/dom@10.4.1) + '@xterm/headless': + specifier: ^6.0.0 + version: 6.0.0 + fake-indexeddb: + specifier: ^6.2.5 + version: 6.2.5 jsdom: specifier: ^30.0.1 version: 30.0.1 @@ -507,9 +516,15 @@ packages: '@xterm/addon-search@0.16.0': resolution: {integrity: sha512-9OeuBFu0/uZJPu+9AHKY6g/w0Czyb/Ut0A5t79I4ULoU4IfU5BEpPFVGQxP4zTTMdfZEYkVIRYbHBX1xWwjeSA==} + '@xterm/addon-serialize@0.14.0': + resolution: {integrity: sha512-uteyTU1EkrQa2Ux6P/uFl2fzmXI46jy5uoQMKEOM0fKTyiW7cSn0WrFenHm5vO5uEXX/GpwW/FgILvv3r0WbkA==} + '@xterm/addon-web-links@0.12.0': resolution: {integrity: sha512-4Smom3RPyVp7ZMYOYDoC/9eGJJJqYhnPLGGqJ6wOBfB8VxPViJNSKdgRYb8NpaM6YSelEKbA2SStD7lGyqaobw==} + '@xterm/headless@6.0.0': + resolution: {integrity: sha512-5Yj1QINYCyzrZtf8OFIHi47iQtI+0qYFPHmouEfG8dHNxbZ9Tb9YGSuLcsEwj9Z+OL75GJqPyJbyoFer80a2Hw==} + '@xterm/xterm@6.0.0': resolution: {integrity: sha512-TQwDdQGtwwDt+2cgKDLn0IRaSxYu1tSUjgKarSDkUM0ZNiSRXFpjxEsvc/Zgc5kq5omJ+V0a8/kIM2WD3sMOYg==} @@ -637,6 +652,10 @@ packages: resolution: {integrity: sha512-KfYbmpRm0VbLjEvVa9yGwCi9GI34xvi7A/HXYWQO65CSD2u3MczUJSuwXKFIxlGsgBQizV9q5J9NHj4VG0n+pA==} engines: {node: '>=12.0.0'} + fake-indexeddb@6.2.5: + resolution: {integrity: sha512-CGnyrvbhPlWYMngksqrSSUT1BAVP49dZocrHuK0SvtR0D5TMs5wP0o3j7jexDJW01KSadjBp1M/71o/KR3nD1w==} + engines: {node: '>=18'} + fdir@6.5.0: resolution: {integrity: sha512-tIbYtZbucOs0BRGqPJkshJUYdL+SDH7dVM8gjy+ERp3WAUjLEFJE+02kanyHtwjWOnwrKYBiwAmM0p4kLJAnXg==} engines: {node: '>=12.0.0'} @@ -1538,8 +1557,12 @@ snapshots: '@xterm/addon-search@0.16.0': {} + '@xterm/addon-serialize@0.14.0': {} + '@xterm/addon-web-links@0.12.0': {} + '@xterm/headless@6.0.0': {} + '@xterm/xterm@6.0.0': {} ansi-regex@5.0.1: {} @@ -1640,6 +1663,8 @@ snapshots: expect-type@1.4.0: {} + fake-indexeddb@6.2.5: {} + fdir@6.5.0(picomatch@4.0.4): optionalDependencies: picomatch: 4.0.4 diff --git a/web/src/Terminal.tsx b/web/src/Terminal.tsx index 533b3e71..5a997c18 100644 --- a/web/src/Terminal.tsx +++ b/web/src/Terminal.tsx @@ -4,6 +4,7 @@ import { Terminal as XTerm } from "@xterm/xterm"; import { FitAddon } from "@xterm/addon-fit"; import { WebLinksAddon } from "@xterm/addon-web-links"; import { SearchAddon, type ISearchResultChangeEvent } from "@xterm/addon-search"; +import { SerializeAddon } from "@xterm/addon-serialize"; import "@xterm/xterm/css/xterm.css"; import { openAttach } from "./api"; import type { RuntimeSocket } from "./runtimeTransport"; @@ -20,13 +21,23 @@ import { loadTerminalCache, MAX_TERMINAL_CACHE_BYTES, saveTerminalCache, + type TerminalCacheEntry, } from "./terminalCache"; import { createReplayQueue, prepareReplayTail, scheduleReplay, + shouldDeferCacheReplay, + snapshotStartPosition, type ReplayHandle, + type ReplayTail, } from "./terminalReplay"; + +// Cap on the scrollback the serialized cache persists. A 5000-line scrollback +// full of wide chars and colour serializes large; 2000 lines keeps the cache +// entry and the one-write restore bounded while still filling the viewport and +// a deep scroll on reload (WI-129). +const SERIALIZE_MAX_SCROLLBACK = 2000; import { beginForegroundReplay } from "./terminalPrewarm"; import { clampTerminalFontSize, @@ -135,12 +146,17 @@ const TerminalView: Component = (props) => { let ws: RuntimeSocket | null = null; let fit: FitAddon | null = null; let search: SearchAddon | null = null; + let serializeAddon: SerializeAddon | null = null; let resizeObserver: ResizeObserver | null = null; let inSnapshot = true; let outputPosition: number | undefined; let snapshotEndPosition: number | undefined; let cacheChunks: Uint8Array[] = []; let cacheBytes = 0; + // A parked pane's cache entry, loaded but not yet restored. Kept in memory on + // reload and restored lazily on first activation (F3), so retained tabs do not + // time-slice the shared replay parser with the active pane. + let deferredCache: TerminalCacheEntry | null = null; let cacheTimer: ReturnType | null = null; // The outputPosition at the last successful persist, so an unchanged ring is // not re-copied and re-written to IndexedDB. @@ -315,6 +331,19 @@ const TerminalView: Component = (props) => { return combined; }; + // Serialize the live terminal's screen + a capped scrollback, so a reload + // restores it in one write instead of re-parsing raw bytes (F5). Returns null + // when serialization is unavailable or throws, so persistCache can fall back + // to the raw ring. + const trySerialize = (): string | null => { + if (!serializeAddon) return null; + try { + return serializeAddon.serialize({ scrollback: SERIALIZE_MAX_SCROLLBACK }); + } catch { + return null; + } + }; + const persistCache = () => { if (outputPosition === undefined) return; // Skip the write when nothing new has arrived since the last persist @@ -322,7 +351,12 @@ const TerminalView: Component = (props) => { // value means the ring is identical and copying+writing it is wasted work. if (outputPosition === lastPersistedPosition) return; lastPersistedPosition = outputPosition; - void saveTerminalCache(props.sessionId, outputPosition, cachedBytes()); + const serialized = trySerialize(); + void saveTerminalCache( + props.sessionId, + outputPosition, + serialized !== null ? { serialized } : { data: cachedBytes() }, + ); }; const scheduleCachePersist = () => { @@ -540,6 +574,8 @@ const TerminalView: Component = (props) => { fit = new FitAddon(); term.loadAddon(fit); term.loadAddon(new WebLinksAddon()); + serializeAddon = new SerializeAddon(); + term.loadAddon(serializeAddon); search = new SearchAddon(); term.loadAddon(search); search.onDidChangeResults((info) => props.onSearchResults?.(info)); @@ -807,36 +843,23 @@ const TerminalView: Component = (props) => { if (destroyed) return; // A live pane's initial load holds the pre-warm gate until snapshot-done. enterForegroundReplay(); - if (!cached || cached.data.byteLength === 0) { + if (!cached) { setReadyToConnect(true); if (!isParked()) connect(); return; } - const bytes = new Uint8Array(cached.data); - const prepared = prepareReplayTail(bytes, cached.outputPosition); - cacheChunks = [bytes]; - cacheBytes = bytes.byteLength; - setStatusText("Restoring terminal..."); - replay = scheduleReplay( - props.sessionId, - [prepared.data], - (chunk, done) => { - if (!term) { - done(); - return; - } - term.write(chunk, done); - }, - { - kind: "cache", - droppedBytes: prepared.droppedBytes, - droppedLines: prepared.droppedLines, - }, - ); - void replay.done.then(() => { + // Adopt the cache's position now so a warm reattach resumes from it even + // if the visible restore is deferred (or later cancelled by a re-park). + outputPosition = cached.outputPosition; + if (shouldDeferCacheReplay(isParked(), true)) { + // Parked on reload: keep the entry in memory and restore it on the first + // activation, not into the shared FIFO with the active pane (F3). + deferredCache = cached; + setReadyToConnect(true); + return; + } + restoreCache(cached, () => { if (destroyed) return; - outputPosition = prepared.outputPosition; - term?.scrollToBottom(); setReadyToConnect(true); if (!isParked()) connect(); }); @@ -895,6 +918,69 @@ const TerminalView: Component = (props) => { connect(); } + /** Replay a prepared cache tail into xterm through the shared replay queue. */ + function replayCacheTail(prepared: ReplayTail): ReplayHandle { + return scheduleReplay( + props.sessionId, + [prepared.data], + (chunk, done) => { + if (!term) { + done(); + return; + } + term.write(chunk, done); + }, + { + kind: "cache", + droppedBytes: prepared.droppedBytes, + droppedLines: prepared.droppedLines, + }, + ); + } + + /** + * Restore a cached terminal, then run `onDone`. A serialized entry (F5) is a + * single `term.write` of the screen + capped scrollback — no raw re-parse. A + * raw entry (pre-warm, fallback, or a pre-F5 cache) re-parses a ground-state + * tail through the shared replay queue, exactly as before F5. + */ + function restoreCache(cached: TerminalCacheEntry, onDone: () => void): void { + setStatusText("Restoring terminal..."); + if (typeof cached.serialized === "string") { + term?.reset(); + cacheChunks = []; + cacheBytes = 0; + outputPosition = cached.outputPosition; + const restored = cached.serialized; + if (!term) { + onDone(); + return; + } + term.write(restored, () => { + if (destroyed) return; + term?.scrollToBottom(); + onDone(); + }); + return; + } + if (cached.data) { + const bytes = new Uint8Array(cached.data); + const prepared = prepareReplayTail(bytes, cached.outputPosition); + cacheChunks = [bytes]; + cacheBytes = bytes.byteLength; + outputPosition = prepared.outputPosition; + replay = replayCacheTail(prepared); + void replay.done.then(() => { + if (destroyed) return; + outputPosition = prepared.outputPosition; + term?.scrollToBottom(); + onDone(); + }); + return; + } + onDone(); + } + function connect() { if (isParked()) return; if (isSessionGone()) { markSessionGone(); return; } @@ -965,9 +1051,13 @@ const TerminalView: Component = (props) => { } if (typeof ctrl.scrollback_pos === "number") { snapshotEndPosition = ctrl.scrollback_pos; - outputPosition = Math.max( - 0, - ctrl.scrollback_pos - (ctrl.scrollback_bytes ?? 0), + // Re-anchor to the start of the payload we are about to replay. + // On a reset:true that followed a resume_from (F1's aged-out or + // over-budget path) this discards the now-stale cursor; the live + // stream after snapshot-done resumes at scrollback_pos with no gap. + outputPosition = snapshotStartPosition( + ctrl.scrollback_pos, + ctrl.scrollback_bytes ?? 0, ); } inSnapshot = true; @@ -999,7 +1089,14 @@ const TerminalView: Component = (props) => { ws?.close(); scheduleReconnect(100); } else if (ctrl.type === "pong") { - if (watchdog.notePong(ctrl.id, ctrl.pos, outputPosition ?? 0) === "recycle") { + if ( + watchdog.notePong( + ctrl.id, + ctrl.pos, + outputPosition ?? 0, + Date.now(), + ) === "recycle" + ) { recycleSocket("server output is ahead of the rendered cursor"); } } @@ -1078,6 +1175,18 @@ const TerminalView: Component = (props) => { if (destroyed || !readyToConnect() || isParked()) return; // Resuming to the foreground: hold the pre-warm gate through this attach. enterForegroundReplay(); + // A tab parked on reload deferred its cache restore (F3); run it now, before + // attaching, so the restored screen is up when the delta arrives. + const pending = deferredCache; + if (pending) { + deferredCache = null; + replay?.cancel(); + restoreCache(pending, () => { + if (destroyed || isParked()) return; + connect(); + }); + return; + } setStatusText("Loading terminal..."); connect(); } diff --git a/web/src/__tests__/attachAuthFrame.test.ts b/web/src/__tests__/attachAuthFrame.test.ts index 211a2fc9..32a72ef9 100644 --- a/web/src/__tests__/attachAuthFrame.test.ts +++ b/web/src/__tests__/attachAuthFrame.test.ts @@ -71,7 +71,7 @@ describe("attach auth frame", () => { expect(frame.resume_from).toBeUndefined(); }); - it("omits the tail hint and sends the cursor on a warm reattach", () => { + it("carries both the cursor and the tail budget on a warm reattach (F1)", () => { installRuntimeTransport(networkless); setBase("http://engine.test"); setToken("test-token-1234567890abcdef"); @@ -79,7 +79,9 @@ describe("attach auth frame", () => { const frame = firstSent(); expect(frame.type).toBe("auth"); expect(frame.resume_from).toBe(4096); - // The tail cap is a cold-attach affordance; a warm reattach must not narrow. - expect(frame.snapshot_tail_bytes).toBeUndefined(); + // F1: the tail budget is sent on every attach. A retained delta within + // budget is still returned byte-exact (reset:false); an aged-out or + // over-budget delta is capped to a bounded reset instead of the whole ring. + expect(frame.snapshot_tail_bytes).toBe(REPLAY_TAIL_MAX_BYTES); }); }); diff --git a/web/src/__tests__/serializeRestore.test.ts b/web/src/__tests__/serializeRestore.test.ts new file mode 100644 index 00000000..628e9487 --- /dev/null +++ b/web/src/__tests__/serializeRestore.test.ts @@ -0,0 +1,71 @@ +import { Terminal } from "@xterm/headless"; +import { SerializeAddon } from "@xterm/addon-serialize"; +import { describe, expect, it } from "vitest"; + +import claudeCodeTui from "../../tests/fixtures/transcripts/claude-code-tui.bin.gz?gzbytes"; +import shellPlain from "../../tests/fixtures/transcripts/shell-plain.bin.gz?gzbytes"; +import cargoBuild from "../../tests/fixtures/transcripts/cargo-build.bin.gz?gzbytes"; + +// F5 (WI-129) persists a live pane's xterm state via @xterm/addon-serialize and +// restores it in a single `term.write` on reload. The fidelity contract is: a +// serialized screen, written into a fresh terminal, reproduces the same screen +// and scrollback — so restore-then-reserialize is a fixed point. Proven here +// against the real transcript corpus (H2), through the same core VT parser the +// browser build uses (@xterm/headless shares it). + +const SERIALIZE_MAX_SCROLLBACK = 2000; + +const CORPORA = [ + { name: "claude-code-tui", bytes: claudeCodeTui }, + { name: "shell-plain", bytes: shellPlain }, + { name: "cargo-build", bytes: cargoBuild }, +] as const; + +function makeTerminal() { + const term = new Terminal({ + cols: 120, + rows: 40, + scrollback: 5000, + allowProposedApi: true, + }); + const serialize = new SerializeAddon(); + term.loadAddon(serialize); + return { term, serialize }; +} + +function write(term: Terminal, data: Uint8Array | string): Promise { + return new Promise((r) => term.write(data, r)); +} + +describe.each(CORPORA)("serialized restore fidelity: $name", ({ bytes }) => { + it("restore-then-reserialize is a fixed point (screen + scrollback preserved)", async () => { + const a = makeTerminal(); + await write(a.term, bytes); + const serialized = a.serialize.serialize({ scrollback: SERIALIZE_MAX_SCROLLBACK }); + // Note: an alt-screen program that has since exited leaves an empty normal + // buffer, so `serialized` can legitimately be "" — the fixed-point below + // still holds, and that is the property F5 relies on. + + // Restore into a fresh terminal in ONE write, then re-serialize. + const b = makeTerminal(); + await write(b.term, serialized); + const reserialized = b.serialize.serialize({ scrollback: SERIALIZE_MAX_SCROLLBACK }); + + expect(reserialized).toBe(serialized); + // The live screen (the viewport) matches too, independent of scrollback. + expect(b.serialize.serialize({ scrollback: 0 })).toBe( + a.serialize.serialize({ scrollback: 0 }), + ); + }); + + it("caps the serialized scrollback to the documented bound", async () => { + const { term, serialize } = makeTerminal(); + await write(term, bytes); + const capped = serialize.serialize({ scrollback: SERIALIZE_MAX_SCROLLBACK }); + // Never more scrollback lines than the cap (rows of viewport aside): the + // serialized string's newline count stays bounded regardless of a + // 5000-line buffer. + const lines = capped.split("\n").length; + expect(lines).toBeLessThanOrEqual(SERIALIZE_MAX_SCROLLBACK + term.rows + 2); + }); +}); diff --git a/web/src/__tests__/shouldDeferCacheReplay.test.ts b/web/src/__tests__/shouldDeferCacheReplay.test.ts new file mode 100644 index 00000000..1e5a1a90 --- /dev/null +++ b/web/src/__tests__/shouldDeferCacheReplay.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, it } from "vitest"; + +import { shouldDeferCacheReplay } from "../terminalReplay"; + +describe("shouldDeferCacheReplay", () => { + it("defers only a parked pane that has cached bytes", () => { + // A retained-but-parked tab on reload keeps its cache in memory and replays + // it lazily on activation, not into the shared FIFO with the active pane. + expect(shouldDeferCacheReplay(true, true)).toBe(true); + }); + + it("does not defer the active pane", () => { + // The pane the user is looking at replays its cache immediately. + expect(shouldDeferCacheReplay(false, true)).toBe(false); + }); + + it("does not defer when there is nothing cached", () => { + // No cache means nothing to replay: a parked pane with an empty cache just + // waits to connect, it does not enter the deferred path. + expect(shouldDeferCacheReplay(true, false)).toBe(false); + expect(shouldDeferCacheReplay(false, false)).toBe(false); + }); +}); diff --git a/web/src/__tests__/snapshotStartPosition.test.ts b/web/src/__tests__/snapshotStartPosition.test.ts new file mode 100644 index 00000000..fbafa53d --- /dev/null +++ b/web/src/__tests__/snapshotStartPosition.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from "vitest"; + +import { snapshotStartPosition } from "../terminalReplay"; + +describe("snapshotStartPosition", () => { + it("is the payload start: end position minus payload length", () => { + // A cold or warm snapshot ending at absolute position 1000 that carries + // 400 bytes started at 600. + expect(snapshotStartPosition(1000, 400)).toBe(600); + }); + + it("clamps at zero when the payload spans the whole stream so far", () => { + // The whole ring is the payload (nothing has aged out): the start is 0, not + // negative. + expect(snapshotStartPosition(400, 400)).toBe(0); + expect(snapshotStartPosition(400, 1000)).toBe(0); + }); + + it("re-anchors a reset that followed a stale resume cursor (F1)", () => { + // The client had rendered up to 2_000_000 and reattached with that cursor. + // The cursor aged out of a 4 MiB ring, so the server sent a bounded tail: + // a 1 MiB payload ending at 5_000_000, reset:true. The client must discard + // its stale cursor and re-anchor to the start of the tail it will replay. + const scrollbackPos = 5_000_000; + const scrollbackBytes = 1_048_576; + const start = snapshotStartPosition(scrollbackPos, scrollbackBytes); + expect(start).toBe(scrollbackPos - scrollbackBytes); + expect(start).toBeGreaterThan(2_000_000); // well past the stale cursor + + // After replaying exactly the payload's bytes, the client's position is the + // snapshot end again, so the live stream resumes with no gap or duplicate. + expect(start + scrollbackBytes).toBe(scrollbackPos); + }); + + it("treats a zero-length snapshot as a no-op at the end position", () => { + expect(snapshotStartPosition(1234, 0)).toBe(1234); + }); +}); diff --git a/web/src/__tests__/terminalCacheRoundtrip.test.ts b/web/src/__tests__/terminalCacheRoundtrip.test.ts new file mode 100644 index 00000000..e29954e9 --- /dev/null +++ b/web/src/__tests__/terminalCacheRoundtrip.test.ts @@ -0,0 +1,76 @@ +import "fake-indexeddb/auto"; +import { beforeEach, describe, expect, it } from "vitest"; + +import { loadTerminalCache, saveTerminalCache } from "../terminalCache"; + +// F5 (WI-129) makes the cache hold either a serialized xterm screen (a live +// pane, the fast restore path) or raw bytes (the headless pre-warm path, or a +// pre-F5 entry). loadTerminalCache must return whichever shape was stored, and +// keep reading raw entries so the format change needs no DB version bump. + +async function clearDb(): Promise { + await new Promise((resolve) => { + const req = indexedDB.deleteDatabase("vogt-terminal-cache"); + req.onsuccess = req.onerror = req.onblocked = () => resolve(); + }); +} + +describe("terminal cache round-trip", () => { + beforeEach(clearDb); + + it("stores and returns a serialized entry (the F5 fast path)", async () => { + await saveTerminalCache("s-serialized", 4242, { serialized: "\x1b[32mrestored\x1b[0m" }); + const entry = await loadTerminalCache("s-serialized"); + expect(entry).not.toBeNull(); + expect(entry?.serialized).toBe("\x1b[32mrestored\x1b[0m"); + expect(entry?.data).toBeUndefined(); + expect(entry?.outputPosition).toBe(4242); + }); + + it("stores and returns a raw entry (pre-warm / fallback / pre-F5)", async () => { + const bytes = new TextEncoder().encode("line-a\nline-b\n"); + await saveTerminalCache("s-raw", bytes.byteLength, { data: bytes }); + const entry = await loadTerminalCache("s-raw"); + expect(entry).not.toBeNull(); + expect(entry?.serialized).toBeUndefined(); + expect(Array.from(new Uint8Array(entry!.data!))).toEqual(Array.from(bytes)); + }); + + it("ground-state-trims a raw entry whose cursor is past its length", async () => { + // outputPosition beyond the byte length marks a ring that dropped its head: + // the load path advances past the first newline so replay starts in ground + // state. + const bytes = new TextEncoder().encode("partial-escape\nclean-line\n"); + await saveTerminalCache("s-trim", 100_000, { data: bytes }); + const entry = await loadTerminalCache("s-trim"); + expect(Array.from(new Uint8Array(entry!.data!))).toEqual( + Array.from(new TextEncoder().encode("clean-line\n")), + ); + }); + + it("ignores an entry with neither serialized nor data", async () => { + // A malformed / foreign entry never crashes the restore; it reads as a cold + // start. + const db = await new Promise((resolve, reject) => { + const req = indexedDB.open("vogt-terminal-cache", 1); + req.onupgradeneeded = () => { + req.result.createObjectStore("sessions", { keyPath: "sessionId" }) + .createIndex("updatedAt", "updatedAt"); + }; + req.onsuccess = () => resolve(req.result); + req.onerror = () => reject(req.error); + }); + await new Promise((resolve, reject) => { + const tx = db.transaction("sessions", "readwrite"); + tx.objectStore("sessions").put({ + sessionId: "s-bad", + outputPosition: 10, + updatedAt: Date.now(), + }); + tx.oncomplete = () => resolve(); + tx.onerror = () => reject(tx.error); + }); + db.close(); + expect(await loadTerminalCache("s-bad")).toBeNull(); + }); +}); diff --git a/web/src/__tests__/terminalPrewarmAttach.test.ts b/web/src/__tests__/terminalPrewarmAttach.test.ts index e4ad4c20..7fe6878a 100644 --- a/web/src/__tests__/terminalPrewarmAttach.test.ts +++ b/web/src/__tests__/terminalPrewarmAttach.test.ts @@ -98,12 +98,13 @@ describe("warmAttachOnce", () => { await expect(promise).resolves.toBe(true); expect(saveTerminalCache).toHaveBeenCalledTimes(1); - const [id, outputPosition, data] = saveTerminalCache.mock.calls[0]!; + const [id, outputPosition, cachePayload] = saveTerminalCache.mock.calls[0]!; expect(id).toBe("11111111-1111-1111-1111-111111111111"); // outputPosition is the absolute stream end, so a later open sends it as // resume_from and gets a delta instead of a cold snapshot. expect(outputPosition).toBe(1000); - expect(data.byteLength).toBe(payload.byteLength); + // Pre-warm has no xterm to serialize, so it caches raw bytes (F5). + expect(cachePayload.data.byteLength).toBe(payload.byteLength); expect(socket.closed).toBe(true); }); diff --git a/web/src/__tests__/terminalWatchdog.test.ts b/web/src/__tests__/terminalWatchdog.test.ts index 4ef00c1f..da05490b 100644 --- a/web/src/__tests__/terminalWatchdog.test.ts +++ b/web/src/__tests__/terminalWatchdog.test.ts @@ -21,16 +21,60 @@ describe("terminal socket watchdog", () => { expect(watchdog.check(100 + WATCHDOG_TIMEOUT_MS)).toBe("healthy"); }); - it("recycles when pong proves that output was missed", () => { + it("does not recycle on a single behind pong; it arms a confirm probe", () => { + // F2: one behind pong is a suspect, not a verdict — the pong can momentarily + // precede the very chunks it is ahead of. No recycle, and check() issues a + // prompt confirm probe rather than waiting a full interval. const watchdog = new SocketWatchdog(); watchdog.notePingSent(7, 1_000); - expect(watchdog.notePong(7, 42, 41)).toBe("recycle"); + expect(watchdog.notePong(7, 42, 41, 1_010)).toBe("healthy"); + expect(watchdog.check(1_020, false)).toBe("probe"); + }); + + it("recycles on a second behind pong within the window with no output", () => { + const watchdog = new SocketWatchdog(); + watchdog.notePingSent(1, 1_000); + expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // first: suspect + watchdog.notePingSent(2, 1_020); // the confirm probe + // Second behind pong, within WATCHDOG_TIMEOUT_MS of the first, nothing in + // between: a real stall. + expect(watchdog.notePong(2, 43, 41, 1_030)).toBe("recycle"); + }); + + it("clears the suspicion when output arrives between the two probes", () => { + const watchdog = new SocketWatchdog(); + watchdog.notePingSent(1, 1_000); + expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // suspect + watchdog.noteOutput(1_015); // the stream is live after all + watchdog.notePingSent(2, 1_020); + // The next behind pong starts a fresh suspicion, it does not recycle. + expect(watchdog.notePong(2, 43, 41, 1_030)).toBe("healthy"); + }); + + it("clears the suspicion when the confirm pong shows the client caught up", () => { + const watchdog = new SocketWatchdog(); + watchdog.notePingSent(1, 1_000); + expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // suspect + watchdog.notePingSent(2, 1_020); + expect(watchdog.notePong(2, 42, 42, 1_030)).toBe("healthy"); // caught up + // Suspicion cleared: the next lone behind pong is again only a suspect. + watchdog.notePingSent(3, 1_040); + expect(watchdog.notePong(3, 50, 41, 1_050)).toBe("healthy"); + }); + + it("drops a stale suspect the confirm never resolved in time", () => { + const watchdog = new SocketWatchdog(); + watchdog.notePingSent(1, 1_000); + expect(watchdog.notePong(1, 42, 41, 1_010)).toBe("healthy"); // suspect at 1010 + // No confirm pong arrives; past the timeout the suspicion is dropped and the + // watchdog falls back to its ordinary interval cadence. + expect(watchdog.check(1_010 + WATCHDOG_TIMEOUT_MS, false)).toBe("healthy"); }); it("does not probe again before the periodic interval", () => { const watchdog = new SocketWatchdog(); watchdog.notePingSent(1, 0); - watchdog.notePong(1, 0, 0); + watchdog.notePong(1, 0, 0, 0); expect(watchdog.check(WATCHDOG_INTERVAL_MS - 1, false)).toBe("healthy"); }); }); diff --git a/web/src/__tests__/transcriptFidelity.test.ts b/web/src/__tests__/transcriptFidelity.test.ts new file mode 100644 index 00000000..a58a54c1 --- /dev/null +++ b/web/src/__tests__/transcriptFidelity.test.ts @@ -0,0 +1,119 @@ +import { describe, expect, it } from "vitest"; + +import claudeCodeTui from "../../tests/fixtures/transcripts/claude-code-tui.bin.gz?gzbytes"; +import shellPlain from "../../tests/fixtures/transcripts/shell-plain.bin.gz?gzbytes"; +import cargoBuild from "../../tests/fixtures/transcripts/cargo-build.bin.gz?gzbytes"; + +import { groundStateReplayStart } from "../terminalCache"; +import { prepareReplayTail, sliceForReplay } from "../terminalReplay"; + +// Real PTY captures (see tests/fixtures/transcripts/README.md). These carry +// genuine escape sequences, SGR colour, cursor moves, alt-screen switches and +// UTF-8, so the ground-state trimming is proven against the byte patterns it +// actually meets — not synthetic `line\n` data. +// +// `lineOriented` corpora emit line feeds (the ground-state seam the trim aligns +// to). `claude-code-tui` is a pure alt-screen redraw stream that addresses the +// cursor directly and never emits `\n` — the case with NO seam, where the +// raw-byte trim can only fall back (and which F5's serialized-state restore, +// WI-129, is meant to cover). Keeping it in the corpus proves the trim stays +// source-exact even then. +const CORPORA = [ + { name: "claude-code-tui", bytes: claudeCodeTui, lineOriented: false }, + { name: "shell-plain", bytes: shellPlain, lineOriented: true }, + { name: "cargo-build", bytes: cargoBuild, lineOriented: true }, +] as const; +const STEP = 4096; + +const isContinuationByte = (b: number | undefined): boolean => + b !== undefined && (b & 0xc0) === 0x80; + +describe.each(CORPORA)("transcript fidelity: $name", ({ bytes: corpus, lineOriented }) => { + it("is a non-trivial real transcript with escape sequences", () => { + expect(corpus.byteLength).toBeGreaterThan(STEP); + expect(corpus.includes(0x1b)).toBe(true); + expect(corpus.includes(0x0a)).toBe(lineOriented); + }); + + it("groundStateReplayStart picks a ground-state, source-exact seam at every 4 KiB cut", () => { + let checked = 0; + // Start at STEP: off === 0 is "nothing dropped" (outputPosition equals the + // length), where a start of 0 is correct with no seam to align. + for (let off = STEP; off < corpus.byteLength; off += STEP) { + // Simulate the client ring dropping its oldest `off` bytes at an arbitrary + // offset (which may split an escape or a UTF-8 char). outputPosition > + // tail.length signals that a drop occurred. + const tail = corpus.subarray(off); + const start = groundStateReplayStart(tail, corpus.byteLength); + + if (start > 0) { + // The only position we can prove is ground state: just past a line feed. + expect(tail[start - 1]).toBe(0x0a); + expect(isContinuationByte(tail[start])).toBe(false); + } else { + // start 0 on a dropped tail means there was no line feed to align to. + expect(tail.indexOf(0x0a)).toBe(-1); + } + // The grounded tail is the exact same bytes of the source from off+start: + // a view over the same buffer at the same offset, never a re-encoding. + const grounded = tail.subarray(start); + const expected = corpus.subarray(off + start); + expect(grounded.byteOffset).toBe(expected.byteOffset); + expect(grounded.byteLength).toBe(expected.byteLength); + checked += 1; + } + expect(checked).toBeGreaterThan(4); + }); + + it("leaves a seamless (alt-screen) tail intact rather than cutting blind", () => { + // Without a line feed there is no position the raw-byte trim can prove is + // ground state, so groundStateReplayStart returns 0 (the tail is replayed + // as-is) instead of guessing a cut inside an escape. Documented limitation + // that F5 addresses by persisting serialized screen state. + if (lineOriented) return; + const tail = corpus.subarray(Math.floor(corpus.byteLength / 2)); + expect(tail.indexOf(0x0a)).toBe(-1); + expect(groundStateReplayStart(tail, corpus.byteLength)).toBe(0); + }); + + it("prepareReplayTail bounds the tail, aligns it, and keeps it source-exact", () => { + for (let off = STEP; off < corpus.byteLength; off += STEP) { + const maxBytes = corpus.byteLength - off; + const prepared = prepareReplayTail(corpus, corpus.byteLength, maxBytes); + + // A tail is a suffix of the corpus (same backing buffer, aligned at the end). + const startAbs = corpus.byteLength - prepared.data.byteLength; + expect(prepared.data.byteOffset).toBe(startAbs); + expect(prepared.data.buffer).toBe(corpus.buffer); + + // Ground state: stream start, or just past a line feed; never a + // continuation byte. + if (startAbs > 0) { + expect(corpus[startAbs - 1]).toBe(0x0a); + } + if (prepared.data.byteLength > 0) { + expect(isContinuationByte(prepared.data[0])).toBe(false); + } + + // Bounded to the budget whenever a newline seam exists at/after the cut; + // a newline-free tail is documented to be left intact. + if (corpus.indexOf(0x0a, off) !== -1) { + expect(prepared.data.byteLength).toBeLessThanOrEqual(maxBytes); + } + } + }); + + it("sliceForReplay never begins inside an escape or a UTF-8 code point", () => { + // A tighter budget forces a real cut on every corpus; the kept tail must + // still begin in ground state. + for (const budget of [16 * 1024, 64 * 1024, 128 * 1024]) { + if (budget >= corpus.byteLength) continue; + const tail = sliceForReplay(corpus, budget); + const startAbs = corpus.byteLength - tail.byteLength; + if (startAbs > 0) { + expect(corpus[startAbs - 1]).toBe(0x0a); + expect(isContinuationByte(tail[0])).toBe(false); + } + } + }); +}); diff --git a/web/src/api.ts b/web/src/api.ts index 06aebe9e..c0a257db 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -1429,13 +1429,15 @@ export function openAttach(id: string, resumeFrom?: number): RuntimeSocket { ws.send(JSON.stringify({ type: "auth", token: tok, - // Cold attach (no cached cursor): bound the server's full snapshot to - // the same tail budget the cache-restore path already enforces, so a - // first open never ships and replays the whole scrollback ring. - // A warm reattach sends its cursor instead and its delta is untouched. - ...(resumeFrom === undefined - ? { snapshot_tail_bytes: REPLAY_TAIL_MAX_BYTES } - : { resume_from: resumeFrom }), + // The tail budget is sent on EVERY attach, warm or cold (F1). It bounds + // the server's reply on all three paths: a cold snapshot, a warm delta + // whose cursor aged out of the ring, and a warm delta that is simply + // larger than the budget — all capped to a ground-state tail the same + // size the cache-restore path enforces, so no attach ever ships and + // replays the whole scrollback ring. A warm delta that fits the budget + // is still returned byte-exact (reset:false). + snapshot_tail_bytes: REPLAY_TAIL_MAX_BYTES, + ...(resumeFrom === undefined ? {} : { resume_from: resumeFrom }), })); }, { once: true }, diff --git a/web/src/gzbytes.d.ts b/web/src/gzbytes.d.ts new file mode 100644 index 00000000..ce469327 --- /dev/null +++ b/web/src/gzbytes.d.ts @@ -0,0 +1,6 @@ +// A `?gzbytes` import (see the `gz-fixture-bytes` plugin in vitest.config.ts) +// resolves to the decoded bytes of a gzip fixture, for tests only. +declare module "*?gzbytes" { + const bytes: Uint8Array; + export default bytes; +} diff --git a/web/src/terminalCache.ts b/web/src/terminalCache.ts index 680aa9ea..0cbbe526 100644 --- a/web/src/terminalCache.ts +++ b/web/src/terminalCache.ts @@ -5,13 +5,34 @@ const MAX_CACHED_SESSIONS = 8; export const MAX_TERMINAL_CACHE_BYTES = 4 * 1024 * 1024; +/** + * A cached terminal, in one of two shapes (F5, WI-129): + * + * - `serialized`: a live pane persists its xterm screen + capped scrollback via + * `@xterm/addon-serialize`. On reload it restores in a single `term.write`, + * with no raw-byte re-parse — the fast path. + * - `data`: raw scrollback bytes. Written by the headless pre-warm path + * (`terminalPrewarm`), which has no xterm to serialize, and by a live pane as + * a fallback when serialization is unavailable. Restored by re-parsing a + * ground-state-aligned tail (the pre-F5 path), which also reads any entry a + * pre-F5 client left behind — so the format change needs no DB version bump + * or migration. + * + * Exactly one of `serialized` / `data` is present. + */ export interface TerminalCacheEntry { sessionId: string; outputPosition: number; - data: ArrayBuffer; updatedAt: number; + serialized?: string; + data?: ArrayBuffer; } +/** What a caller hands `saveTerminalCache`: a serialized screen or raw bytes. */ +export type TerminalCachePayload = + | { serialized: string } + | { data: Uint8Array }; + /** * The client cache is a byte-oriented ring (see `appendToCache` in * Terminal.tsx): once it overflows, its oldest bytes are dropped at an @@ -73,22 +94,30 @@ export async function loadTerminalCache( >, ); db.close(); + if (!result || !Number.isSafeInteger(result.outputPosition)) return null; + + // Serialized fast path: a live pane's screen, restored in one write. + if (typeof result.serialized === "string") { + return result.serialized.length > 0 ? result : null; + } + // Raw path (pre-warm entries, live-pane fallback, pre-F5 entries). Detect + // the ArrayBuffer realm-safely (`instanceof` misses a cross-realm buffer, as + // a structured clone can produce). if ( - !result || - !(result.data instanceof ArrayBuffer) || - !Number.isSafeInteger(result.outputPosition) || - result.outputPosition < result.data.byteLength + result.data != null && + Object.prototype.toString.call(result.data) === "[object ArrayBuffer]" && + result.outputPosition >= result.data.byteLength ) { - return null; - } - // Drop any partial leading escape sequence / UTF-8 char left by the ring - // trim so the tail replays from a terminal ground state. - const bytes = new Uint8Array(result.data); - const start = groundStateReplayStart(bytes, result.outputPosition); - if (start > 0) { - result.data = bytes.slice(start).buffer; + // Drop any partial leading escape sequence / UTF-8 char left by the ring + // trim so the tail replays from a terminal ground state. + const bytes = new Uint8Array(result.data); + const start = groundStateReplayStart(bytes, result.outputPosition); + if (start > 0) { + result.data = bytes.slice(start).buffer; + } + return result; } - return result; + return null; } catch { return null; } @@ -97,7 +126,7 @@ export async function loadTerminalCache( export async function saveTerminalCache( sessionId: string, outputPosition: number, - data: Uint8Array, + payload: TerminalCachePayload, ): Promise { if (typeof indexedDB === "undefined" || !Number.isSafeInteger(outputPosition)) { return; @@ -106,12 +135,11 @@ export async function saveTerminalCache( const db = await openCache(); const tx = db.transaction(STORE_NAME, "readwrite"); const store = tx.objectStore(STORE_NAME); - store.put({ - sessionId, - outputPosition, - data: data.slice().buffer, - updatedAt: Date.now(), - } satisfies TerminalCacheEntry); + const stored: TerminalCacheEntry = + "serialized" in payload + ? { sessionId, outputPosition, serialized: payload.serialized, updatedAt: Date.now() } + : { sessionId, outputPosition, data: payload.data.slice().buffer, updatedAt: Date.now() }; + store.put(stored); const entries = await requestResult( store.index("updatedAt").getAllKeys() as IDBRequest, diff --git a/web/src/terminalPrewarm.ts b/web/src/terminalPrewarm.ts index b72c72e4..049a4dc6 100644 --- a/web/src/terminalPrewarm.ts +++ b/web/src/terminalPrewarm.ts @@ -212,7 +212,9 @@ export function warmAttachOnce( return; } const data = concatChunks(chunks, bytes); - void saveTerminalCache(sessionId, endPosition, data).finally(() => + // Pre-warm has no xterm to serialize; it caches raw bytes, restored + // via the ground-state-aligned raw path (F5). + void saveTerminalCache(sessionId, endPosition, { data }).finally(() => finish(true), ); } diff --git a/web/src/terminalReplay.ts b/web/src/terminalReplay.ts index 388dd01f..fe1c56e5 100644 --- a/web/src/terminalReplay.ts +++ b/web/src/terminalReplay.ts @@ -11,6 +11,46 @@ export const REPLAY_TAIL_MAX_BYTES = 1 * 1024 * 1024; /** Keep each xterm parser turn bounded and aligned with the server frame size. */ export const REPLAY_SLICE_BYTES = 64 * 1024; +/** + * On reload, should a pane defer its cached-scrollback replay? + * + * A parked pane — a retained tab that is not the active one, or the unfocused + * half of a split — must NOT replay its cache into the shared replay FIFO on + * reload: N retained tabs replaying at once time-slice the single parser N ways + * and starve the pane the user is actually looking at. A parked pane keeps its + * cache in memory and replays lazily on first activation (`resumeSocket`) + * instead. An active pane with a cache replays immediately. + */ +export function shouldDeferCacheReplay( + parked: boolean, + hasCachedBytes: boolean, +): boolean { + return parked && hasCachedBytes; +} + +/** + * The absolute output position at the START of a snapshot payload: the byte + * offset the first snapshot byte sits at. The server reports the position at + * the END of the snapshot (`scrollback_pos`) and the payload's byte length + * (`scrollback_bytes`); the start is their difference, clamped at zero. + * + * This is the cursor a client adopts when a `snapshot-start` arrives, and it is + * what makes F1's bounded reset safe. When a `reset:true` snapshot follows a + * `resume_from` — the cursor aged out of the ring, or the delta exceeded the + * budget — the server sends a ground-state tail whose start is + * `scrollback_pos - scrollback_bytes`, well after the client's now-stale + * cursor. The client discards that stale cursor and re-anchors here, so once it + * has replayed the `scrollback_bytes` of the tail its position is exactly + * `scrollback_pos` again and the live stream resumes with no gap and no + * duplicate. + */ +export function snapshotStartPosition( + scrollbackPos: number, + scrollbackBytes: number, +): number { + return Math.max(0, scrollbackPos - scrollbackBytes); +} + export interface ReplayTail { /** The ground-state-safe tail to send to xterm. */ data: Uint8Array; diff --git a/web/src/terminalWatchdog.ts b/web/src/terminalWatchdog.ts index 993c6b3d..75deed0a 100644 --- a/web/src/terminalWatchdog.ts +++ b/web/src/terminalWatchdog.ts @@ -14,16 +14,43 @@ export class SocketWatchdog { private pending: PendingPing | null = null; private lastProbeAt = Number.NEGATIVE_INFINITY; private lastOutputAt = Number.NEGATIVE_INFINITY; + // A single "behind" pong (server position ahead of what the client has + // rendered) is a suspect, not a verdict: even with the engine answering a + // probe with its actual sent position, a pong can momentarily precede the + // very chunks it is ahead of. So the first behind pong arms a prompt confirm + // probe instead of recycling; only a second behind pong within the timeout, + // with no output in between, is treated as a real stall. `suspectAt` is when + // the first behind pong was seen, or null when there is no live suspicion. + private suspectAt: number | null = null; notePingSent(id: number, at: number): void { this.pending = { id, at }; this.lastProbeAt = at; } - notePong(id: number, serverPos: number, localPos: number): WatchdogResult { + notePong( + id: number, + serverPos: number, + localPos: number, + at: number, + ): WatchdogResult { if (!this.pending || this.pending.id !== id) return "healthy"; this.pending = null; - return serverPos > localPos ? "recycle" : "healthy"; + if (serverPos <= localPos) { + // Caught up (or the engine reported its true sent position): any prior + // suspicion is cleared. + this.suspectAt = null; + return "healthy"; + } + // Behind. Recycle only on a second behind pong within the timeout window + // that no output cleared in between (a genuine stall); otherwise arm a + // suspect and let `check` issue a prompt confirm probe. + if (this.suspectAt !== null && at - this.suspectAt < WATCHDOG_TIMEOUT_MS) { + this.suspectAt = null; + return "recycle"; + } + this.suspectAt = at; + return "healthy"; } noteOutput(at: number): void { @@ -31,6 +58,9 @@ export class SocketWatchdog { // Old engines do not answer the probe. Output after it is still proof of // life, so an idle shell does not reconnect in a loop against one. if (this.pending && at >= this.pending.at) this.pending = null; + // Output flowing is proof the stream is live, so a behind-pong suspicion is + // no longer credible. + this.suspectAt = null; } check(now: number, forceProbe = false): WatchdogResult { @@ -42,6 +72,16 @@ export class SocketWatchdog { } return "recycle"; } + // A live suspect confirms promptly — one more probe now, not a full + // interval later — so a real stall is caught within the timeout. A suspect + // the confirm never resolved in time is dropped as stale. + if (this.suspectAt !== null) { + if (now - this.suspectAt >= WATCHDOG_TIMEOUT_MS) { + this.suspectAt = null; + } else { + return "probe"; + } + } return forceProbe || now - this.lastProbeAt >= WATCHDOG_INTERVAL_MS ? "probe" : "healthy"; @@ -51,5 +91,6 @@ export class SocketWatchdog { this.pending = null; this.lastProbeAt = Number.NEGATIVE_INFINITY; this.lastOutputAt = Number.NEGATIVE_INFINITY; + this.suspectAt = null; } } diff --git a/web/tests/browser/terminalReplay.spec.ts b/web/tests/browser/terminalReplay.spec.ts new file mode 100644 index 00000000..eae3005f --- /dev/null +++ b/web/tests/browser/terminalReplay.spec.ts @@ -0,0 +1,290 @@ +// Terminal replay budget + live acceptance (H3, WI-124). +// +// Two projects, one file: +// +// - The default mocked projects (`desktop`/`phone`, Vite dev server) stream a +// bounded corpus into a real terminal over a routed WebSocket and assert the +// per-pane replay stays under budget and actually renders. No live stack. +// +// - The `live` project (only registered when PLAYWRIGHT_LIVE_BASE_URL is set, +// selected by e2e.yml) drives a real load session on a real stack and asserts +// the two operator symptoms directly: switching away and back replays only a +// bounded tail (F1; a no-reset resume once F4 lands), and a reload does not +// slow the active pane (F3/F5), with no spurious `[disconnected]` (F2). See +// docs/local/TERMINAL_ATTACH_BUDGET_PLAN.md. + +import { readFileSync } from "node:fs"; +import { gunzipSync } from "node:zlib"; + +import { expect, test, type Page } from "@playwright/test"; + +const isLive = (project: string) => project === "live"; + +// ─── Mocked project ──────────────────────────────────────────────────────── + +// ~1 MiB — the client's per-pane replay budget (REPLAY_TAIL_MAX_BYTES), i.e. the +// most a cold attach delivers once F1 bounds it. Built from a real transcript. +function buildCorpus(target: number): Uint8Array { + const seed = gunzipSync( + readFileSync(new URL("../fixtures/transcripts/shell-plain.bin.gz", import.meta.url)), + ); + const out = new Uint8Array(target); + for (let off = 0; off < target; off += seed.byteLength) { + out.set(seed.subarray(0, Math.min(seed.byteLength, target - off)), off); + } + return out; +} +const CORPUS = buildCorpus(1024 * 1024); + +// The CI per-pane replay budget. Generous headroom over the measured parse rate +// (~5–6 MB/s ⇒ ~0.2 s for 1 MiB) so it is a real ceiling, not a flaky stopwatch; +// tune down once the self-hosted runner's number is known. +const REPLAY_BUDGET_MS = 2500; + +const SESSION = { + id: "sess-load", + name: "load", + cwd: "/workspace", + activity: "idle", + exit_code: null, + scrollback_bytes: 0, + created_at: "2026-09-09T00:00:00Z", +}; + +async function mockedFixtures(page: Page): Promise { + await page.addInitScript(() => { + localStorage.setItem("vogt.token", "browser-test-token"); + localStorage.setItem("vogt.appTheme.v1", "dark"); + }); + const json = (body: unknown) => (route: { fulfill: (o: unknown) => Promise }) => + route.fulfill({ json: body }); + await page.route("**/api/install/status", json({ install_mode: false })); + await page.route( + "**/api/auth/check", + json({ + ok: true, + version: "test", + product_version: "test", + storage: { state_dir: "/tmp", workspace_root: "/workspace" }, + }), + ); + await page.route( + "**/api/status**", + json({ + version: "test", + session_count: 1, + push_subscription_count: 0, + gui_process_count: 0, + gui_stream_configured: false, + fcm_enabled: false, + history: { + enabled: true, + archived_session_count: 0, + log_file_count: 0, + log_bytes: 0, + db_bytes: 0, + }, + agent_tasks: { + task_count: 0, + prompt_task_dir_count: 0, + prompt_file_count: 0, + context_file_count: 0, + prompt_bytes: 0, + orphan_task_dir_count: 0, + }, + auth_broker: { auto_agent_auth: false, helper: "disabled" }, + storage: { state_dir: "/tmp", workspace_root: "/workspace" }, + }), + ); + await page.route( + "**/api/config**", + json({ + assistant_enabled: false, + gui_stream_url: null, + session_templates: [], + gui_stream_available: false, + vogt: { configured: true }, + }), + ); + await page.route("**/api/sessions", (route) => route.fulfill({ json: [SESSION] })); + await page.route("**/api/sessions/*", (route) => { + if (route.request().method() !== "GET") return route.fulfill({ json: { ok: true } }); + return route.fulfill({ json: { ...SESSION, scrollback_base64: "" } }); + }); + await page.route("**/api/events", (route) => + route.fulfill({ + status: 200, + contentType: "text/event-stream", + body: `data: ${JSON.stringify({ type: "activity", id: "s", state: "idle" })}\n\n`, + }), + ); +} + +// Route the attach WebSocket and stream `corpus` as one cold snapshot, exactly +// as the engine would (snapshot-start → binary chunks → snapshot-done). +async function streamSnapshot(page: Page, corpus: Uint8Array): Promise { + await page.routeWebSocket(/\/api\/sessions\/.*\/attach/, (ws) => { + ws.onMessage(() => { + ws.send( + JSON.stringify({ + type: "snapshot-start", + session_id: SESSION.id, + scrollback_bytes: corpus.byteLength, + scrollback_pos: corpus.byteLength, + reset: true, + }), + ); + const CHUNK = 64 * 1024; + for (let i = 0; i < corpus.byteLength; i += CHUNK) { + ws.send(Buffer.from(corpus.subarray(i, Math.min(i + CHUNK, corpus.byteLength)))); + } + ws.send(JSON.stringify({ type: "snapshot-done" })); + }); + }); +} + +test.describe("terminal replay budget (mocked)", () => { + test.beforeEach(({}, testInfo) => { + test.skip(isLive(testInfo.project.name), "mocked only"); + }); + + test("a bounded snapshot renders and replays under the per-pane budget", async ({ page }) => { + const replayLines: string[] = []; + page.on("console", (msg) => { + if (msg.text().includes("[vogt] terminal replay")) replayLines.push(msg.text()); + }); + + await mockedFixtures(page); + await streamSnapshot(page, CORPUS); + await page.goto("/#/t/sess-load"); + + // The snapshot replay emits a `vogt-terminal-replay.snapshot` measure. + const handle = await page.waitForFunction( + () => { + const m = performance + .getEntriesByType("measure") + .find((e) => e.name === "vogt-terminal-replay.snapshot"); + return m ? { duration: m.duration } : null; + }, + undefined, + { timeout: 20_000 }, + ); + const measure = (await handle.jsonValue()) as { duration: number }; + + // The measure only exists once the replay has drained into xterm, so its + // presence proves the 1 MiB snapshot rendered; its duration is the per-pane + // budget the plan bounds. + expect(measure.duration).toBeGreaterThan(0); + expect(measure.duration).toBeLessThan(REPLAY_BUDGET_MS); + + // The client logged the `[vogt] terminal replay` telemetry the live spec + // reads (kind, snapshotBytes, replayDurationMs). + await expect.poll(() => replayLines.length, { timeout: 5_000 }).toBeGreaterThan(0); + }); +}); + +// ─── Live project ────────────────────────────────────────────────────────── +// +// Runs only under PLAYWRIGHT_LIVE_BASE_URL (the `live` project), against a real +// stack booted by e2e.yml with a load-generating session. It asserts the two +// operator symptoms as the plan's H3 requires. Not exercised by mocked runs. + +const LIVE_TOKEN = process.env.PLAYWRIGHT_LIVE_TOKEN ?? ""; +// ~3 MiB/min of coloured, sequence-numbered output, then idle — the same shape +// as the engine harness's load session, so the ring wraps within the test. +const LIVE_LOAD_COMMAND = [ + "/bin/bash", + "-c", + "i=0; while [ \"$i\" -lt 400000 ]; do " + + "printf '\\033[3%dmSEQ%08d the quick brown fox jumps over the lazy dog\\033[0m\\n' " + + '"$((i % 8))" "$i"; i=$((i + 1)); ' + + "if [ \"$((i % 64))\" -eq 0 ]; then printf '\\033[H'; fi; " + + "done; exec sleep 3600", +]; + +async function liveCreateLoadSession(page: Page): Promise { + const res = await page.request.post("/api/sessions", { + headers: { authorization: `Bearer ${LIVE_TOKEN}` }, + data: { name: "h3-load", command: LIVE_LOAD_COMMAND }, + }); + expect(res.ok()).toBeTruthy(); + return ((await res.json()) as { id: string }).id; +} + +async function liveScrollbackPos(page: Page, id: string): Promise { + const res = await page.request.get(`/api/sessions/${id}`, { + headers: { authorization: `Bearer ${LIVE_TOKEN}` }, + }); + return ((await res.json()) as { scrollback_pos?: number }).scrollback_pos ?? 0; +} + +test.describe("terminal replay budget (live acceptance)", () => { + test.beforeEach(({}, testInfo) => { + test.skip(!isLive(testInfo.project.name), "live only"); + }); + + test("symptom 1: switch away past the ring and back — bounded reattach, no disconnect", async ({ + page, + }) => { + const replays: { kind?: string; snapshotBytes?: number; reset?: boolean }[] = []; + page.on("console", (msg) => { + const text = msg.text(); + if (!text.includes("[vogt] terminal replay")) return; + const m = text.match(/\{.*\}/); + if (m) { + try { + replays.push(JSON.parse(m[0])); + } catch { + /* not the structured arg */ + } + } + }); + + const id = await liveCreateLoadSession(page); + await page.goto(`/#/t/${id}`); + await expect(page.locator(".xterm-rows")).toContainText("SEQ", { timeout: 20_000 }); + const openPos = await liveScrollbackPos(page, id); + + // Switch away to another place, wait until the ring has wrapped well past + // where we were, then switch back. + await page.goto("/#/sessions"); + await expect + .poll(() => liveScrollbackPos(page, id), { timeout: 60_000, intervals: [500] }) + .toBeGreaterThan(openPos + 4 * 1024 * 1024); + replays.length = 0; // only measure the reattach + await page.goto(`/#/t/${id}`); + await expect(page.locator(".xterm-rows")).toContainText("SEQ", { timeout: 20_000 }); + + // No [disconnected] marker in the buffer. + await expect(page.locator(".xterm-rows")).not.toContainText("[disconnected]"); + // The reattach replayed at most the budget (F1). It may be a bounded reset + // until F4 lands, when it becomes a no-reset resume; either way it is <= budget. + const reattach = replays.filter((r) => r.kind === "snapshot" || r.kind === "cache"); + for (const r of reattach) { + expect(r.snapshotBytes ?? 0).toBeLessThanOrEqual(1024 * 1024); + } + }); + + test("symptom 2: reload with the session cached — active pane under budget", async ({ page }) => { + const id = await liveCreateLoadSession(page); + await page.goto(`/#/t/${id}`); + await expect(page.locator(".xterm-rows")).toContainText("SEQ", { timeout: 20_000 }); + // Let the client persist its cache, then reload. + await page.waitForTimeout(6000); + await page.reload(); + + const handle = await page.waitForFunction( + () => { + const m = performance + .getEntriesByType("measure") + .find((e) => e.name.startsWith("vogt-terminal-replay.")); + return m ? { duration: m.duration } : null; + }, + undefined, + { timeout: 20_000 }, + ); + const measure = (await handle.jsonValue()) as { duration: number }; + expect(measure.duration).toBeLessThan(REPLAY_BUDGET_MS); + await expect(page.locator(".xterm-rows")).not.toContainText("[disconnected]"); + }); +}); diff --git a/web/tests/fixtures/transcripts/README.md b/web/tests/fixtures/transcripts/README.md new file mode 100644 index 00000000..9d43c0d4 --- /dev/null +++ b/web/tests/fixtures/transcripts/README.md @@ -0,0 +1,57 @@ +# Terminal transcript corpus + +Real PTY byte streams used by the parser-fidelity tests +(`src/__tests__/transcriptFidelity.test.ts`) and, later, the browser replay +budget spec (H3, WI-124) and the serialized-restore fidelity compare (F5, +WI-129). They exist so the ground-state trimming that both the server ring and +the client cache rely on is proven against **real** escape sequences, SGR +colour, cursor moves, alt-screen switches and UTF-8 — not synthetic `line\n` +data. + +Each file is gzip-compressed (`.bin.gz`); the tests `gunzip` them in memory. + +| Fixture | Source | Exercises | +|---|---|---| +| `claude-code-tui.bin.gz` | a curses TUI redraw loop under a PTY | alt-screen enter/leave, cursor addressing, SGR colour, OSC title, box-drawing + emoji UTF-8 — the shape of an agent TUI | +| `shell-plain.bin.gz` | `ls -la --color=always -R` + UTF-8 text under a PTY | coloured `ls` output, prompts, `\r`, multibyte box/CJK/emoji characters | +| `cargo-build.bin.gz` | a real `cargo build -v` (clean local crates) | green-bold `Compiling`/`Fresh` status lines, `\r`, long verbose rustc command lines with paths | + +## How these were captured + +The **intended production source** is a real dev-stack session: + +``` +GET /api/history/:id/download # raw recorded PTY bytes for a session +``` + +An agent session cannot reach a running stack (its token has no engine +`sessions` capability — see the terminal-attach plan), so the checked-in +fixtures were captured from real programs on a developer box with +`scripts/capture_transcript.py` (a `pty.fork` capture), for example: + +```sh +python3 scripts/capture_transcript.py tui.raw -- python3 tui.py # a curses redraw loop +python3 scripts/capture_transcript.py shell.raw -- bash -lc 'ls -la --color=always -R /usr/include; …' +python3 scripts/capture_transcript.py cargo.raw -- bash -lc 'cargo clean -p … ; cargo build -v --color always' +``` + +Every capture is then run through the sanitiser, which rewrites home paths, +tokens, JWTs and email addresses to inert placeholders **and asserts that no +secret pattern survives** before it is committed: + +```sh +python3 scripts/sanitise_transcript.py tui.raw tui.bin +gzip -9 tui.bin # -> claude-code-tui.bin.gz +``` + +To refresh from a real stack, download a transcript, run it through +`sanitise_transcript.py --check` (it must pass), sanitise, gzip, and replace the +file here — keeping the same three shapes. + +## Sizes + +These local stand-ins are smaller than a full-session download (the TUI and +`cargo` captures in particular): the fidelity tests sample every 4 KiB, so a few +hundred KB per corpus already gives dozens of independent cut offsets across +real escape/UTF-8 boundaries. Replace them with larger real-session downloads +when a stack is available. diff --git a/web/tests/fixtures/transcripts/cargo-build.bin.gz b/web/tests/fixtures/transcripts/cargo-build.bin.gz new file mode 100644 index 00000000..eefcfc8f Binary files /dev/null and b/web/tests/fixtures/transcripts/cargo-build.bin.gz differ diff --git a/web/tests/fixtures/transcripts/claude-code-tui.bin.gz b/web/tests/fixtures/transcripts/claude-code-tui.bin.gz new file mode 100644 index 00000000..c5513107 Binary files /dev/null and b/web/tests/fixtures/transcripts/claude-code-tui.bin.gz differ diff --git a/web/tests/fixtures/transcripts/shell-plain.bin.gz b/web/tests/fixtures/transcripts/shell-plain.bin.gz new file mode 100644 index 00000000..cae7f0b2 Binary files /dev/null and b/web/tests/fixtures/transcripts/shell-plain.bin.gz differ diff --git a/web/vitest.config.ts b/web/vitest.config.ts index 365c2e2e..77eb685c 100644 --- a/web/vitest.config.ts +++ b/web/vitest.config.ts @@ -10,11 +10,42 @@ // and required for the tests, and one file that means two things depending on // who loaded it is how a build starts differing from what was tested. +import { gunzipSync } from "node:zlib"; +import { readFileSync } from "node:fs"; +import { dirname, resolve } from "node:path"; + import { defineConfig } from "vitest/config"; import solid from "vite-plugin-solid"; +// Import a gzip fixture as a decoded `Uint8Array` from a browser-typed test: +// `import bytes from "./x.bin.gz?gzbytes"`. The gunzip happens here in Node +// (this config is not part of the app's browser-only tsconfig), so the tests +// need no `@types/node` and stay free of Node built-ins. The bytes are emitted +// as base64 and decoded with `atob` (a DOM global jsdom provides). +function gzFixtureBytes() { + const SUFFIX = "?gzbytes"; + return { + name: "gz-fixture-bytes", + resolveId(id: string, importer: string | undefined) { + if (!id.endsWith(SUFFIX)) return null; + const file = id.slice(0, -SUFFIX.length); + const base = importer ? dirname(importer) : process.cwd(); + return resolve(base, file) + SUFFIX; + }, + load(id: string) { + if (!id.endsWith(SUFFIX)) return null; + const file = id.slice(0, -SUFFIX.length); + const base64 = gunzipSync(readFileSync(file)).toString("base64"); + return ( + `export default Uint8Array.from(atob(${JSON.stringify(base64)}), ` + + `(c) => c.charCodeAt(0));` + ); + }, + }; +} + export default defineConfig({ - plugins: [solid()], + plugins: [solid(), gzFixtureBytes()], resolve: { // Solid ships two builds. `browser` is the one with a real DOM renderer, // and `development` is the one that keeps the reactive graph's dev