mim/tools/terminal_e2e.py

#!/usr/bin/env python3
"""PTY-backed terminal E2E harness for mim.

Runs a tiny high-signal scenario through a real pseudo-terminal and writes
browser-viewable terminal artifacts. No third-party dependencies.
"""
from __future__ import annotations

import argparse
import hashlib
import html
import os
import pty
import re
import select
import signal
import shutil
import sys
import tempfile
import time
from pathlib import Path

ESC = 0x1B
SCENARIO_ID = "raw-key-chrome-crlf"
TMP_RE = re.compile(r"/tmp/mim-terminal-e2e-[^/]+")


def visible_controls(data: bytes) -> str:
    out: list[str] = []
    text = data.decode("utf-8", errors="replace")
    for ch in text:
        code = ord(ch)
        if ch == "\r":
            out.append("␍")
        elif ch == "\n":
            out.append("␊\n")
        elif ch == "\x1b":
            out.append("␛")
        elif ch == "\t":
            out.append("⇥")
        elif code < 0x20:
            out.append(f"␀{code:02x}")
        else:
            out.append(ch)
    return "".join(out)


def normalize_visible_controls(text: str) -> str:
    text = TMP_RE.sub("<TMP>", text)
    text = re.sub(r"/opt/repositories/mim/zig-out/bin/mim", "<MIM>", text)
    return text


def has_bare_lf(data: bytes) -> bool:
    return any(byte == 0x0A and (i == 0 or data[i - 1] != 0x0D) for i, byte in enumerate(data))


def self_test() -> None:
    sample = b"a\nb\r\n\x1b[31m\t\x01"
    visible = visible_controls(sample)
    expected = "a␊\nb␍␊\n␛[31m⇥␀01"
    if visible != expected:
        raise SystemExit(f"visible control self-test failed: {visible!r} != {expected!r}")
    if not has_bare_lf(b"a\nb"):
        raise SystemExit("bare LF self-test failed: did not detect bare LF")
    if has_bare_lf(b"a\r\nb"):
        raise SystemExit("bare LF self-test failed: flagged CRLF")


class TerminalGrid:
    def __init__(self, width: int, height: int) -> None:
        self.width = width
        self.height = height
        self.rows = [[" " for _ in range(width)] for _ in range(height)]
        self.row = 0
        self.col = 0

    def clear(self) -> None:
        self.rows = [[" " for _ in range(self.width)] for _ in range(self.height)]
        self.row = 0
        self.col = 0

    def newline(self) -> None:
        self.row += 1
        if self.row >= self.height:
            self.rows.pop(0)
            self.rows.append([" " for _ in range(self.width)])
            self.row = self.height - 1

    def put_char(self, ch: str) -> None:
        if ch == "\r":
            self.col = 0
            return
        if ch == "\n":
            self.newline()
            return
        if self.col >= self.width:
            self.col = 0
            self.newline()
        if 0 <= self.row < self.height and 0 <= self.col < self.width:
            self.rows[self.row][self.col] = ch
        self.col += 1

    def feed(self, data: bytes) -> None:
        i = 0
        text = data.decode("utf-8", errors="replace")
        while i < len(text):
            ch = text[i]
            if ch == "\x1b" and i + 1 < len(text) and text[i + 1] == "[":
                j = i + 2
                while j < len(text) and not ("@" <= text[j] <= "~"):
                    j += 1
                if j >= len(text):
                    break
                command = text[j]
                params = text[i + 2 : j]
                if command == "H":
                    self.row = 0
                    self.col = 0
                elif command == "J" and params.endswith("2"):
                    self.clear()
                # SGR and unknown CSI are intentionally ignored for semantic grid.
                i = j + 1
                continue
            self.put_char(ch)
            i += 1

    def lines(self) -> list[str]:
        return ["".join(row).rstrip() for row in self.rows]


def read_available(fd: int, timeout: float = 0.12) -> bytes:
    chunks: list[bytes] = []
    deadline = time.monotonic() + timeout
    while True:
        remaining = max(0.0, deadline - time.monotonic())
        ready, _, _ = select.select([fd], [], [], remaining)
        if not ready:
            break
        try:
            chunk = os.read(fd, 65536)
        except OSError:
            break
        if not chunk:
            break
        chunks.append(chunk)
        deadline = time.monotonic() + 0.04
    return b"".join(chunks)


def spawn_under_pty(argv: list[str], width: int, height: int) -> tuple[int, int]:
    pid, fd = pty.fork()
    if pid == 0:
        os.environ["TERM"] = "xterm-256color"
        try:
            import fcntl
            import struct
            import termios

            winsize = struct.pack("HHHH", height, width, 0, 0)
            fcntl.ioctl(0, termios.TIOCSWINSZ, winsize)
        except Exception:
            pass
        os.execv(argv[0], argv)
    return pid, fd


def svg_for_lines(lines: list[str], width: int, height: int) -> str:
    cell_w = 9
    cell_h = 18
    pad = 12
    svg_w = pad * 2 + width * cell_w
    svg_h = pad * 2 + height * cell_h
    body = []
    for idx, line in enumerate(lines[:height]):
        body.append(
            f'<text x="{pad}" y="{pad + (idx + 1) * cell_h}" xml:space="preserve">{html.escape(line)}</text>'
        )
    return "\n".join(
        [
            f'<svg xmlns="http://www.w3.org/2000/svg" width="{svg_w}" height="{svg_h}" viewBox="0 0 {svg_w} {svg_h}">',
            '<rect width="100%" height="100%" fill="#12161e"/>',
            '<style>text{font-family:ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;font-size:14px;fill:#d6deeb}</style>',
            *body,
            "</svg>",
        ]
    )


def html_for_lines(lines: list[str], raw_svg_name: str) -> str:
    escaped = html.escape("\n".join(lines))
    return f"""<!doctype html>
<meta charset="utf-8">
<title>mim terminal-e2e</title>
<style>
body {{ margin: 0; background: #0f131a; color: #d6deeb; }}
.terminal {{ white-space: pre; font: 14px/18px ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; padding: 12px; }}
a {{ color: #7ee787; }}
</style>
<div class="terminal">{escaped}</div>
<p><a href="{html.escape(raw_svg_name)}">SVG screenshot artifact</a></p>
"""


def write_artifacts(out_dir: Path, transcript: bytes, lines: list[str], width: int, height: int, saved: str | None) -> None:
    visible = visible_controls(transcript)
    normalized_visible = normalize_visible_controls(visible)
    out_dir.mkdir(parents=True, exist_ok=True)
    (out_dir / "raw.bin").write_bytes(transcript)
    (out_dir / "visible-controls.txt").write_text(visible, encoding="utf-8")
    (out_dir / "visible-controls.normalized.txt").write_text(normalized_visible, encoding="utf-8")
    (out_dir / "transcript.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
    (out_dir / "screenshot.svg").write_text(svg_for_lines(lines, width, height), encoding="utf-8")
    (out_dir / "terminal.html").write_text(html_for_lines(lines, "screenshot.svg"), encoding="utf-8")
    snapshot = [
        f"scenario\t{SCENARIO_ID}",
        f"viewport\t{width}x{height}",
        f"raw_sha256\t{hashlib.sha256(transcript).hexdigest()}",
        f"bare_lf\t{str(has_bare_lf(transcript)).lower()}",
        f"saved_sha256\t{hashlib.sha256((saved or '').encode('utf-8')).hexdigest()}",
        "artifacts\traw.bin visible-controls.txt visible-controls.normalized.txt transcript.txt screenshot.svg terminal.html manifest.tsv",
    ]
    (out_dir / "manifest.tsv").write_text("key\tvalue\n" + "\n".join(snapshot) + "\n", encoding="utf-8")


def run_scenario(mim: Path, out_dir: Path, width: int, height: int) -> None:
    self_test()
    if out_dir.exists():
        shutil.rmtree(out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix="mim-terminal-e2e-") as tmp:
        target = Path(tmp) / "note.txt"
        target.write_text("", encoding="utf-8")
        pid, fd = spawn_under_pty([str(mim), str(target)], width, height)
        transcript = bytearray()
        grid = TerminalGrid(width, height)
        try:
            for payload in [b"", b"w", b"h", b"a", b"t", b" ", b"i", b"s", b" ", b"u", b"p", b" ", b"n", b" ", b"w", b" ", b"q"]:
                if payload:
                    os.write(fd, payload)
                chunk = read_available(fd)
                transcript.extend(chunk)
                grid.feed(chunk)
            deadline = time.monotonic() + 2.0
            while time.monotonic() < deadline:
                done_pid, status = os.waitpid(pid, os.WNOHANG)
                chunk = read_available(fd, 0.03)
                transcript.extend(chunk)
                grid.feed(chunk)
                if done_pid == pid:
                    if status != 0:
                        raise SystemExit(f"mim exited non-zero: status={status}")
                    break
            else:
                write_artifacts(out_dir, bytes(transcript), grid.lines(), width, height, None)
                os.kill(pid, signal.SIGTERM)
                raise SystemExit(f"mim did not exit after save+quit script; artifacts={out_dir}")
        finally:
            try:
                os.close(fd)
            except OSError:
                pass

        saved = target.read_text(encoding="utf-8")
        lines = grid.lines()
        raw = bytes(transcript)
        write_artifacts(out_dir, raw, lines, width, height, saved)

        failures: list[str] = []
        if saved != "what is up":
            failures.append(f"saved file mismatch: {saved!r}")
        joined = "\n".join(lines)
        raw_text = raw.decode("utf-8", errors="ignore")
        if has_bare_lf(raw):
            failures.append("raw terminal output contains bare LF; expected CRLF in raw mode")
        if "1│" not in raw_text:
            failures.append("line-number gutter missing from terminal output")
        if "▌" not in raw_text:
            failures.append("cursor marker missing from terminal output")
        if "mode:normal" not in raw_text and "saved" not in raw_text:
            failures.append("expected editor status not observed")
        if "^[" in raw_text:
            failures.append("escape appears echoed as ^[")
        if "mode:normal" not in raw_text:
            failures.append("Space-n did not leave insert mode before save/quit")
        if any(len(line) > width for line in lines):
            failures.append("terminal line exceeds viewport width")
        if failures:
            print(f"artifacts\t{out_dir}", file=sys.stderr)
            for failure in failures:
                print(f"FAIL\t{failure}", file=sys.stderr)
            raise SystemExit(1)

    print("scenario\tstatus\tartifacts")
    print(f"raw-key-chrome-crlf\tPASS\t{out_dir}")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Run mim through a PTY and write browser-viewable terminal E2E artifacts.",
        epilog=(
            "Golden policy: investigate before updating artifacts. Semantic snapshots "
            "(manifest.tsv, transcript.txt, visible-controls.normalized.txt) are the "
            "primary review surface; screenshot.svg/terminal.html are visual evidence. "
            "It is safe to update artifacts only after explaining intentional UI/control-byte changes."
        ),
    )
    parser.add_argument("mim", type=Path, help="path to compiled mim binary")
    parser.add_argument("--out", type=Path, default=Path(f".zig-cache/terminal-e2e/{SCENARIO_ID}"))
    parser.add_argument("--width", type=int, default=48)
    parser.add_argument("--height", type=int, default=16)
    args = parser.parse_args()
    run_scenario(args.mim.resolve(), args.out, args.width, args.height)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())