test(web): H2 real transcript corpus + parser-fidelity tests (WI-123)

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:]))

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:]))

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);
+      }
+    }
+  });
+});

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;
+}