slhx agent
418 lines
17 KiB
Python
Executable File
418 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""PTY-backed terminal E2E matrix for mim.
|
|
|
|
Runs high-signal scenarios through a real pseudo-terminal and writes
|
|
browser-viewable terminal artifacts. No third-party dependencies.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import dataclasses
|
|
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
|
|
from typing import Callable
|
|
|
|
ESC = 0x1B
|
|
TMP_RE = re.compile(r"/tmp/mim-terminal-e2e-[^/]+")
|
|
CSI_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class Scenario:
|
|
id: str
|
|
width: int
|
|
height: int
|
|
file_state: str
|
|
input_profile: str
|
|
mode_surface: str
|
|
color_mode: str
|
|
target_kind: str
|
|
setup: Callable[[Path], tuple[Path, str | None]]
|
|
payloads: tuple[bytes, ...]
|
|
expect_saved: str | None
|
|
required_raw: tuple[str, ...]
|
|
forbidden_raw: tuple[str, ...] = ("^[",)
|
|
|
|
|
|
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 strip_csi(text: str) -> str:
|
|
return CSI_RE.sub("", text)
|
|
|
|
|
|
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()
|
|
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, color_mode: str, cwd: Path) -> tuple[int, int]:
|
|
pid, fd = pty.fork()
|
|
if pid == 0:
|
|
os.chdir(cwd)
|
|
os.environ["TERM"] = {"truecolor": "xterm-256color", "mono": "vt100"}.get(color_mode, "xterm-256color")
|
|
os.environ["COLORTERM"] = "truecolor" if color_mode == "truecolor" else ""
|
|
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, scenario: Scenario) -> str:
|
|
escaped = html.escape("\n".join(lines))
|
|
meta = html.escape(f"{scenario.id} {scenario.width}x{scenario.height} {scenario.file_state} {scenario.input_profile} {scenario.mode_surface} {scenario.color_mode}")
|
|
return f"""<!doctype html>
|
|
<meta charset="utf-8">
|
|
<title>mim terminal-e2e {html.escape(scenario.id)}</title>
|
|
<style>
|
|
body {{ margin: 0; background: #0f131a; color: #d6deeb; }}
|
|
.meta {{ color: #7ee787; font: 12px ui-monospace, monospace; padding: 8px 12px 0; }}
|
|
.terminal {{ white-space: pre; font: 14px/18px ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; padding: 12px; }}
|
|
a {{ color: #7ee787; }}
|
|
</style>
|
|
<div class="meta">{meta}</div>
|
|
<div class="terminal">{escaped}</div>
|
|
<p><a href="{html.escape(raw_svg_name)}">SVG screenshot artifact</a></p>
|
|
"""
|
|
|
|
|
|
def write_artifacts(out_dir: Path, scenario: Scenario, transcript: bytes, lines: list[str], 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, scenario.width, scenario.height), encoding="utf-8")
|
|
(out_dir / "terminal.html").write_text(html_for_lines(lines, "screenshot.svg", scenario), encoding="utf-8")
|
|
snapshot = [
|
|
f"scenario\t{scenario.id}",
|
|
f"viewport\t{scenario.width}x{scenario.height}",
|
|
f"file_state\t{scenario.file_state}",
|
|
f"input_profile\t{scenario.input_profile}",
|
|
f"mode_surface\t{scenario.mode_surface}",
|
|
f"color_mode\t{scenario.color_mode}",
|
|
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 setup_empty(tmp: Path) -> tuple[Path, str | None]:
|
|
target = tmp / "empty.txt"
|
|
target.write_text("", encoding="utf-8")
|
|
return target, ""
|
|
|
|
|
|
def setup_existing(tmp: Path) -> tuple[Path, str | None]:
|
|
target = tmp / "existing.txt"
|
|
target.write_text("alpha\nbeta\n", encoding="utf-8")
|
|
return target, "alpha\nbeta\n"
|
|
|
|
|
|
def setup_new(tmp: Path) -> tuple[Path, str | None]:
|
|
return tmp / "new-file.txt", ""
|
|
|
|
|
|
def setup_directory(tmp: Path) -> tuple[Path, str | None]:
|
|
root = tmp / "repo"
|
|
root.mkdir()
|
|
(root / "one.zig").write_text("pub fn main() void {}\n", encoding="utf-8")
|
|
return root, None
|
|
|
|
|
|
MOBILE_SAVE_QUIT = tuple(bytes([b]) for b in b"what is up n w q")
|
|
ATTACHED_SAVE_QUIT = (b"\x1b[F", b"!", b" ", b"n", b" ", b"w", b" ", b"q")
|
|
PANEL_QUIT = (b" ", b"n", b" ", b"q")
|
|
|
|
IOS_DEFAULT_KEYS = set(b"abcdefghijklmnopqrstuvwxyz ")
|
|
|
|
SCENARIOS = [
|
|
Scenario("raw-key-chrome-crlf", 48, 16, "empty", "ios-default-qwertz-space-path", "insert-normal", "truecolor", "file", setup_empty, MOBILE_SAVE_QUIT, "what is up", ("1│", "▌", "mode:normal")),
|
|
Scenario("attached-existing-medium", 72, 18, "existing", "attached-keyboard", "insert-normal", "truecolor", "file", setup_existing, ATTACHED_SAVE_QUIT, "alpha!\nbeta", ("1│", "2│", "▌")),
|
|
Scenario("new-file-mono-narrow", 40, 12, "new", "ios-default-qwertz-space-path", "insert-normal", "mono", "file", setup_new, MOBILE_SAVE_QUIT, "what is up", ("1│", "▌", "mode:normal")),
|
|
Scenario("directory-panel-narrow", 52, 12, "directory", "ios-default-qwertz-space-path", "panel", "mono", "directory", setup_directory, PANEL_QUIT, None, ("file", "one.zig")),
|
|
]
|
|
|
|
|
|
def assert_input_profile_is_ergonomic(scenario: Scenario) -> None:
|
|
if scenario.input_profile != "ios-default-qwertz-space-path":
|
|
return
|
|
for payload in scenario.payloads:
|
|
for byte in payload:
|
|
if byte not in IOS_DEFAULT_KEYS:
|
|
raise SystemExit(
|
|
f"{scenario.id}: iOS default keyboard scenario uses non-default key byte 0x{byte:02x}; "
|
|
"move it to an attached-keyboard scenario or redesign the mobile path"
|
|
)
|
|
|
|
|
|
def run_one(mim: Path, root_out: Path, scenario: Scenario) -> tuple[str, Path]:
|
|
assert_input_profile_is_ergonomic(scenario)
|
|
out_dir = root_out / scenario.id
|
|
with tempfile.TemporaryDirectory(prefix="mim-terminal-e2e-") as tmp_name:
|
|
tmp = Path(tmp_name)
|
|
target, expected_initial = scenario.setup(tmp)
|
|
_ = expected_initial
|
|
pid, fd = spawn_under_pty([str(mim), target.name], scenario.width, scenario.height, scenario.color_mode, tmp)
|
|
transcript = bytearray()
|
|
grid = TerminalGrid(scenario.width, scenario.height)
|
|
try:
|
|
for payload in (b"", *scenario.payloads):
|
|
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:
|
|
write_artifacts(out_dir, scenario, bytes(transcript), grid.lines(), None)
|
|
raise SystemExit(f"{scenario.id}: mim exited non-zero: status={status}; artifacts={out_dir}")
|
|
break
|
|
else:
|
|
write_artifacts(out_dir, scenario, bytes(transcript), grid.lines(), None)
|
|
os.kill(pid, signal.SIGTERM)
|
|
raise SystemExit(f"{scenario.id}: mim did not exit after script; artifacts={out_dir}")
|
|
finally:
|
|
try:
|
|
os.close(fd)
|
|
except OSError:
|
|
pass
|
|
|
|
saved: str | None = None
|
|
if scenario.target_kind == "file" and target.exists():
|
|
saved = target.read_text(encoding="utf-8")
|
|
lines = grid.lines()
|
|
raw = bytes(transcript)
|
|
write_artifacts(out_dir, scenario, raw, lines, saved)
|
|
failures: list[str] = []
|
|
raw_text = raw.decode("utf-8", errors="ignore")
|
|
if scenario.expect_saved is not None and saved != scenario.expect_saved:
|
|
failures.append(f"saved file mismatch: {saved!r} != {scenario.expect_saved!r}")
|
|
if has_bare_lf(raw):
|
|
failures.append("raw terminal output contains bare LF; expected CRLF in raw mode")
|
|
plain_text = strip_csi(raw_text)
|
|
for required in scenario.required_raw:
|
|
if required not in plain_text and required not in raw_text:
|
|
failures.append(f"required terminal text missing: {required!r}")
|
|
if scenario.color_mode == "truecolor" and "\x1b[38;2;" not in raw_text:
|
|
failures.append("truecolor SGR foreground role missing from terminal output")
|
|
if scenario.expect_saved and scenario.expect_saved in plain_text and f"▌{scenario.expect_saved}" in plain_text:
|
|
failures.append("cursor marker is pinned before typed content instead of following cursor column")
|
|
if scenario.expect_saved and scenario.expect_saved.replace("\n", "") in plain_text:
|
|
compact_saved = scenario.expect_saved.replace("\n", "")
|
|
if f"▌{compact_saved}" in plain_text:
|
|
failures.append("cursor marker is pinned before typed content instead of following cursor column")
|
|
for forbidden in scenario.forbidden_raw:
|
|
if forbidden in raw_text:
|
|
failures.append(f"forbidden terminal text present: {forbidden!r}")
|
|
if scenario.mode_surface != "panel" and "mode:normal" not in raw_text:
|
|
failures.append("scenario never reached normal mode")
|
|
if any(len(line) > scenario.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{scenario.id}\t{failure}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
return scenario.id, out_dir
|
|
|
|
|
|
def run_matrix(mim: Path, out_root: Path, only: set[str] | None) -> None:
|
|
self_test()
|
|
selected = [scenario for scenario in SCENARIOS if only is None or scenario.id in only]
|
|
unknown = (only or set()) - {scenario.id for scenario in SCENARIOS}
|
|
if unknown:
|
|
raise SystemExit(f"unknown scenario id(s): {', '.join(sorted(unknown))}")
|
|
if out_root.exists():
|
|
shutil.rmtree(out_root)
|
|
out_root.mkdir(parents=True, exist_ok=True)
|
|
rows = ["scenario\tstatus\tartifacts"]
|
|
matrix_rows = ["scenario\tviewport\tfile_state\tinput_profile\tmode_surface\tcolor_mode\tartifacts"]
|
|
for scenario in selected:
|
|
scenario_id, artifact_dir = run_one(mim, out_root, scenario)
|
|
rows.append(f"{scenario_id}\tPASS\t{artifact_dir}")
|
|
matrix_rows.append(f"{scenario.id}\t{scenario.width}x{scenario.height}\t{scenario.file_state}\t{scenario.input_profile}\t{scenario.mode_surface}\t{scenario.color_mode}\t{artifact_dir}")
|
|
(out_root / "matrix.tsv").write_text("\n".join(matrix_rows) + "\n", encoding="utf-8")
|
|
print("\n".join(rows))
|
|
|
|
|
|
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 "
|
|
"(matrix.tsv, 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(".zig-cache/terminal-e2e"))
|
|
parser.add_argument("--scenario", action="append", help="run only this scenario id; may be repeated")
|
|
args = parser.parse_args()
|
|
run_matrix(args.mim.resolve(), args.out, set(args.scenario) if args.scenario else None)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|