feat(movie): simplify the Windows terminal recorder to serve/run/key/watch/close

Windows has no tmux, so the recorder had grown into a 738-line daemon with a
file-based request protocol, request IDs, wait-only result retrieval, and a
Win32 Job Object module. Replace it with the Unix route's shape: serve keeps
ttyd and a headless browser alive and logs the terminal's output; run, key,
watch, and close are one-shot CDP calls against that browser. The installed
prompt reports each command's status through the window title, so run can
print it without any visible marker.

Process cleanup uses taskkill /T (a pgrep walk on Unix) instead of Job
Objects, which also simplifies the card renderer. The session tests run on
macOS too, since nothing in the script is Windows-specific.

Verified: 9 session tests per shell on Windows 11 for PowerShell 5.1,
PowerShell 7, and Git Bash; the browser suite with Chrome and Edge; 44
portable tests on macOS against a real ttyd session.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

docs/superpowers/specs/2026-09-09-proof-movie-windows-completion-design.md

@@ -26,43 +26,29 @@ The narration change addresses observed failures: native Windows could synthesiz
 
 ## 2. Provide a usable Windows terminal example
 
-Add `skills/proving-it-works-with-a-movie/examples/film-terminal.py` as the Windows recording entry point referenced by the Windows instructions. It serves the shell through ttyd and observes it in a headless browser. The existing Unix tmux recipe remains documented separately; do not advertise this new entry point as a tested Unix recorder.
-
-### Session and control contract
-
-- `serve --shell powershell51|powershell7|gitbash --directory <new-session-dir>` owns one recorded shell and one browser page. Allow explicit shell/browser/ttyd executable paths and `--cwd` (default: the invoking working directory). Launch ttyd with explicit `-w` for that cwd, writable mode, one client, and loopback binding.
-- Keep `serve` in a foreground/background task owned by the invoking harness, as the visual companion does. Later shell tool calls control the same process. It is not an installed service or self-daemonizing launcher.
-- `request --directory <session-dir> --file <UTF-8-JSON>` submits a request. Files avoid inline command length and cross-shell JSON quoting problems. A session has one requesting controller; IDs are consecutive integers starting at one. Expose `next_request_id` in the ready/status files and `inspect` result; publish the updated status before acknowledging a consumed ID. Reject duplicate or non-next IDs before publication, with an immediate error rather than waiting for a missing file. Writes and replies are atomic. Default output is acknowledgment JSON, which does not claim command completion; a rejection is nonzero. `--wait-result` waits up to `--timeout` (default: 30 seconds) and exits nonzero for command/control failure, unknown/interrupted outcome, or missing result. A client wait timeout does not cancel the pending command.
-- `result --directory <session-dir> --id <accepted-id> --timeout <seconds>` waits for or reads an existing result without submitting another request or consuming an ID. Default timeout is 30 seconds; use the same completion/exit-status rules as `request --wait-result`. This is the supported follow-up after acknowledgment or client timeout.
-- Supported request operations are `run` (`command`, optional `timeout_seconds`, optional `native_producer`), `begin-take` (`name`), `end-take`, `key` (`key`), `inspect`, `close`, and `cancel`. Support printable keys, Enter, Escape, arrows, Tab, and Ctrl-C. Diagnostic probe phases, crash fixtures, and aggregation are test-only.
-- `run` acknowledges acceptance without waiting for the command to finish. Its result deadline defaults to 90 seconds and is independent of the client's wait timeout. Use numbered request/ack/result files under `control/`, with session ID and request ID in replies. `inspect` exposes pending request and active take. A second `run` while one is pending fails before typing, leaving the session usable; take controls and intentional `key` input remain available. A known failing command can be followed by another command; it does not itself terminate recording.
-- `end-take` stops capture only. It preserves the shell, working directory, variables, pending command, and browser connection for a later take, including across separate harness tool calls.
-- Readiness requires a nonce, shell identity, and cwd returned through the filmed terminal, plus a readable preflight image. Opening another ttyd connection is not an observation mechanism. A reconnect fails the session instead of silently replacing its identity.
-
-Command results distinguish `completed`, `unknown`, and `interrupted`; shell success is nullable. Preserve raw PowerShell `$?` separately from observed request-attributed errors, including PS5.1 expression-wrapper behavior; instrumentation must not change the submitted statement's semantics. Capture status before logging.
-
-`native_producer` identifies the native executable for a direct invocation or the first stage of a pipeline followed by logging commands. This is the bounded attribution contract: Bash uses the first saved `PIPESTATUS` entry; PowerShell uses that request's native exit status. Opaque/mixed scripts omit this field and report their observed shell outcome without claiming individual internal commands succeeded. Do not infer producer identity or reuse stale native status. A completed result is successful only when shell success is true, no request-attributed shell error is present, and any explicitly identified producer has a known zero exit code. Thus successful `tee`/`Tee-Object` cannot hide a producer failure.
-
-### Capture choice and deliberately limited geometry
-
-Use **bounded `Page.captureScreenshot` PNG capture at 5 fps**, with a fixed **1600 × 900** browser viewport set before terminal navigation. This reuses the successful live-screenshot mechanism and the imported skill's deliberate-frame route. Continuous CDP screencast delivery is not required for this port.
-
-- Observe and record actual terminal rows/columns; require at least 80 columns for the existing 60-character completion framing. Do not offer resize controls. A detected terminal geometry change fails the session with an actionable message.
-- The CDP receiver must continue handling terminal output, command replies, and cancellation during capture. Use a two-second screenshot response deadline; a timeout or lost browser fails the active take.
-- Treat 5 fps as the target sampling cadence, not a guarantee that every 0.2-second event is observed. Record monotonic take boundaries and screenshot request/completion times. A take starts at its first completed screenshot; `begin-take` completes only then. Place frames on the 0.2-second output grid using the latest screenshot completed at or before that grid time, never a future screenshot. Record duplicated intervals; a gap between completed captures, or from the last capture to take end, exceeding two seconds fails the take. Match total take duration within one frame (0.2 seconds). Total duration alone does not establish event timing.
-- `end-take` returns a take directory and the corresponding existing `kind: frames` scene fields (`src`, `rate: 5`). These artifacts feed the existing assembler directly.
-- Check preflight pixels and final exported frames. A static application can legitimately produce identical frames; image similarity alone is not a stalled-capture test. Acceptance must show several successive visible TUI states, each held for at least 1.3 seconds, in the automatically captured sequence while the command owns input, before its exit key. Check their order and placement against capture times. A manually requested snapshot cannot rescue this acceptance check; missing states fail it.
-- At the fixed geometry, long wrapping output followed by a command completion must work. Invalid/truncated completion records time out as unknown; they never produce success. A general VT emulator and arbitrary resizing are outside this delivery.
-
-### Ownership and failure handling
-
-Use a Windows Job Object: establish containment before recorder-owned roots execute, and retain kill-on-close semantics. Normal close, cancel, browser loss, and forced recorder exit must remove its browser/ttyd/shell descendants within ten seconds. An unrelated process must survive. Cleanup still runs when capture, control-file, or report writes fail; such failures remain nonzero results. Pending command success becomes unknown/interrupted when observation is lost.
-
-A command-result deadline expiring records an unknown outcome, fails the active take, and terminates the session with owned-resource cleanup. It must not clear pending state and type another command into the still-running program. Browser loss, capture timeout, or a forbidden geometry change likewise ends the session. Client wait deadlines have no such effect.
-
-`close` finalizes a healthy active take; `cancel` marks an active take incomplete. Both mark any pending command interrupted with null success and release owned processes. Earlier completed takes remain available. A shutdown acknowledgment means only accepted; publish the successful close/cancel result only after cleanup finishes. That success describes resource release, not the interrupted command. A finalization, cleanup, or result-write failure makes shutdown nonzero or leaves no successful result; the client must not infer success from a missing result.
-
-Keep ownership code specific to this recorder and its launched browser. Share a small Windows helper between the recorder and the card renderer; do not introduce a generic process framework.
+Windows has no tmux, so `examples/film-terminal.py` stands in for it, keeping
+the Unix route's shape: ttyd serves the shell, a headless Chrome or Edge page
+renders it, screenshots are the frames. `serve` starts both, keeps them alive
+in a harness-owned background task, and appends the raw terminal output to a
+log. `run`, `key`, `watch`, and `close` are one-shot CDP calls against that
+browser, so the shell, its variables, and its cwd persist across separate
+tool calls with no daemon protocol.
+
+- The installed prompt reports a counter, the shell's success flag, the last
+  native exit code, and the cwd through the window title, which the picture
+  never shows. `run` types a command, waits for the next prompt, and prints
+  that status as JSON; it exits 1 when the command failed and 2 when it is
+  still running after `--seconds`.
+- Capture is bounded `Page.captureScreenshot` at 5 fps in a fixed 1600×900
+  viewport. Frames sit on the 0.2-second grid and a slow screenshot repeats
+  the previous frame, so every `--record` directory is a `kind: frames` scene
+  at `rate: 5` for the assembler.
+- `serve` refuses a blank canvas at readiness and exits nonzero if the
+  browser or ttyd connection drops. `close` kills ttyd, the browser, and
+  their descendants (`taskkill /T` on Windows) and removes the browser
+  profile.
+- The script does not check which OS it runs on, which is how its session
+  tests run on macOS too; the Unix route stays the tmux recipe.
 
 ## 3. Update only the Windows-facing guidance
 
@@ -80,6 +66,7 @@ Each of PowerShell 5.1, PowerShell 7, and Git Bash on Windows 11 x64 produced
 a complete narrated, hard-subtitled movie from a title card, a still, real
 browser clicks, two terminal takes across separate tool calls, and a source
 movie segment, using local Piper narration verified by local transcription,
-with the checker passing. The portable suites pass on macOS and were also run
-on Linux during development. Full-desktop `gdigrab` capture returned only
+with the checker passing. The recorder's session tests pass on Windows 11 for
+all three shells and on macOS against a real ttyd; the media suites were also
+run on Linux during development. Full-desktop `gdigrab` capture returned only
 wallpaper on the test host; window-title capture worked.

skills/proving-it-works-with-a-movie/assembling.md

@@ -107,8 +107,8 @@ is small and always worth committing.
 
 Use native `uv`, FFmpeg and ffprobe on the test process's PATH. Hard subtitles
 require FFmpeg's `subtitles` filter (libass). Install Chrome or Edge for cards.
-The tools' Python environments are managed by uv; the media scripts require
-Python 3.10+, and the Windows terminal example requires 3.12+. First use can
+The tools' Python environments are managed by uv and need Python 3.10+.
+First use can
 download Python, script dependencies, the local Piper voice, and the local
 transcription model. Do that setup before recording. No cloud key is required.
 PowerShell needs neither Git Bash nor WSL, tmux, Docker, or administrator rights.

skills/proving-it-works-with-a-movie/examples/film-terminal.py

@@ -1,738 +1,523 @@
 #!/usr/bin/env -S uv run --script
 # /// script
-# requires-python = ">=3.12"
-# dependencies = ["websocket-client==1.9.0"]
+# requires-python = ">=3.10"
+# dependencies = ["websocket-client==1.9.0", "pillow"]
 # ///
-"""Record one persistent native Windows terminal as timed PNG takes."""
-import bisect
-import math
-
-
-def frame_sources(completed, start, end, rate=5):
-    if not completed or completed[0] != start or end < start:
-        raise ValueError("Invalid take boundaries")
-    if completed != sorted(completed) or completed[-1] > end:
-        raise ValueError("Invalid capture times")
-    intervals = [b - a for a, b in zip(completed, completed[1:])]
-    if max(intervals + [end - completed[-1]]) > 2:
-        raise TimeoutError("Capture gap exceeds two seconds")
-    count = max(1, math.ceil((end - start) * rate))
-    return [bisect.bisect_right(completed, start + i / rate) - 1
-            for i in range(count)]
-
-
-def command_succeeded(result):
-    return (result.get("outcome") == "completed"
-            and result.get("shell_success") is True
-            and result.get("shell_error") is None
-            and (result.get("native_producer") is None
-                 or result.get("producer_exit_code") == 0))
-
-
+"""Film a shell on native Windows, where there is no tmux.
+
+ttyd serves the shell over HTTP and a headless Chrome or Edge page renders
+it: the same picture the Unix route in recording-a-terminal.md gets. `serve`
+stands in for tmux: it holds the only ttyd client open so the shell survives
+between tool calls, and appends the raw terminal output to
+SESSION/terminal.log. Every other verb is one short CDP call against that
+browser.
+
+  serve SESSION --shell powershell51|powershell7|gitbash [--cwd DIR]
+        hold the session open; run it in a background task
+  run   SESSION 'command' [--record OUT] [--seconds 60] [--hold 1.5]
+        type the command, film until the prompt returns, print its status
+  key   SESSION Enter|Escape|Tab|Ctrl-C|ArrowDown|q [--record OUT]
+        press one key
+  watch SESSION --record OUT [--seconds 30]
+        film without typing: a TUI after a key, or the tail of long work
+  close SESSION
+        kill ttyd, the browser, and everything they started
+
+The prompt `serve` installs reports each command's status through the
+window title, which the picture never shows, so `run` can print it.
+"""
 import argparse
 import base64
-import codecs
+import io
 import json
 import os
 import re
 import shutil
-import shlex
 import socket
+import subprocess
 import sys
 import time
 import urllib.request
-import uuid
 from pathlib import Path
-sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
-from browser_tools import find_browser
-from windows_jobs import WindowsJob
-
-def file_handoff(operation):
-    # A Windows reader can overlap an atomic replacement. Retry the operation,
-    # never substitute old/empty evidence; persistent access errors still fail.
-    deadline = time.monotonic() + 0.5
-    while True:
-        try:
-            return operation()
-        except PermissionError:
-            if time.monotonic() >= deadline:
-                raise
-            time.sleep(0.01)
 
-def read_json(path):
-    return json.loads(file_handoff(lambda: path.read_text(encoding="utf-8")))
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
+from browser_tools import find_browser, kill_process_tree  # noqa: E402
+
+FPS = 5
+WIDTH, HEIGHT = 1600, 900
+# Title set by the installed prompt: MOVIE;<count>;<ok>;<native exit>;<cwd>
+MARKER = re.compile(rb"\x1b\][012];MOVIE;(\d+);([01]);(-?\d*);([^\x07\x1b]*)(?:\x07|\x1b\\)")
+SHELLS = {
+    "powershell51": (["-NoLogo", "-NoProfile", "-NoExit"], "powershell"),
+    "powershell7": (["-NoLogo", "-NoProfile", "-NoExit"], "pwsh"),
+    "gitbash": (["--noprofile", "--norc", "-i"], "bash"),
+    "bash": (["--noprofile", "--norc", "-i"], "bash"),
+}
+PROMPTS = {
+    "powershell": r'''$global:MovieN = 0
+function global:prompt {
+    $ok = $?; $native = $global:LASTEXITCODE; $global:MovieN++
+    $Host.UI.RawUI.WindowTitle = "MOVIE;$global:MovieN;$([int]$ok);$native;$PWD"
+    "PS $PWD> "
+}
+Clear-Host''',
+    "bash": r'''MOVIE_N=0
+movie_prompt() { local s=$?; MOVIE_N=$((MOVIE_N + 1)); printf '\033]0;MOVIE;%s;%s;%s;%s\a' "$MOVIE_N" "$((s == 0))" "$s" "$PWD"; }
+PROMPT_COMMAND=movie_prompt
+PS1='\w \$ '
+clear''',
+}
+KEYS = {
+    "Enter": dict(key="Enter", code="Enter", windowsVirtualKeyCode=13, text="\r"),
+    "Escape": dict(key="Escape", code="Escape", windowsVirtualKeyCode=27),
+    "Tab": dict(key="Tab", code="Tab", windowsVirtualKeyCode=9, text="\t"),
+    "ArrowLeft": dict(key="ArrowLeft", code="ArrowLeft", windowsVirtualKeyCode=37),
+    "ArrowUp": dict(key="ArrowUp", code="ArrowUp", windowsVirtualKeyCode=38),
+    "ArrowRight": dict(key="ArrowRight", code="ArrowRight", windowsVirtualKeyCode=39),
+    "ArrowDown": dict(key="ArrowDown", code="ArrowDown", windowsVirtualKeyCode=40),
+    "Ctrl-C": dict(key="c", code="KeyC", windowsVirtualKeyCode=67, modifiers=2),
+}
 
-def write_json(path, value):
-    temporary = path.with_suffix(".tmp")
-    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
-    file_handoff(lambda: temporary.replace(path))
-
-def ttyd_output(event: dict) -> bytes:
-    frame = event["params"]["response"]
-    payload = frame["payloadData"]
-    raw = base64.b64decode(payload) if frame["opcode"] == 2 else payload.encode("utf-8")
-    return raw[1:] if raw[:1] == b"0" else b""
-
-class CompletionParser:
-    """Parse bounded multiline records; this is not a VT redraw emulator."""
-
-    def __init__(self):
-        self.decoder = codecs.getincrementaldecoder("utf-8")("replace")
-        self.text = ""
-        self.records = []
-
-    def feed(self, data):
-        self.text = (self.text + self.decoder.decode(data))[-131072:]
-        clean = re.sub(r"\x1b\][^\x07]*(?:\x07|\x1b\\)|\x1b\[[0-?]*[ -/]*[@-~]", "", self.text)
-        for match in re.finditer(r"\[MOVIE\|([A-Za-z0-9+/=\s]+)\|END\]", clean):
-            try:
-                record = json.loads(base64.b64decode(re.sub(r"\s", "", match[1]), validate=True).decode("utf-8"))
-            except (ValueError, UnicodeError):
-                continue
-            if isinstance(record, dict) and record not in self.records:
-                self.records.append(record)
 
-def free_port():
-    with socket.socket() as sock:
-        sock.bind(("127.0.0.1", 0))
-        return sock.getsockname()[1]
+def shell_family(kind):
+    return "bash" if kind in ("bash", "gitbash") else "powershell"
+
+
+def shell_argv(kind, explicit=None):
+    flags, default = SHELLS[kind]
+    exe = explicit
+    if not exe and kind == "gitbash":
+        # PATH may hold WSL's bash.exe; only Git's is a native Windows shell.
+        for root in (os.environ.get("ProgramFiles"), os.environ.get("ProgramW6432")):
+            if root and (Path(root) / "Git/bin/bash.exe").is_file():
+                exe = str(Path(root) / "Git/bin/bash.exe")
+    exe = exe or shutil.which(default)
+    if not exe:
+        raise SystemExit(f"cannot find the {kind} executable; pass --shell-exe")
+    return [str(Path(exe).resolve()), *flags]
+
+
+def prompt_script(kind, cwd):
+    """Enter the cwd (ttyd's own -w is unreliable), install the status prompt, clear."""
+    if shell_family(kind) == "bash":
+        quoted = "'" + str(cwd).replace("\\", "/").replace("'", "'\\''") + "'"
+        return f"cd -- {quoted}\n" + PROMPTS["bash"]
+    quoted = "'" + str(cwd).replace("'", "''") + "'"
+    return f"Set-Location -LiteralPath {quoted}\n" + PROMPTS["powershell"]
+
+
+def prompt_command(kind, cwd):
+    """One typed line that runs prompt_script without any quoting hazards."""
+    encoded = base64.b64encode(prompt_script(kind, cwd).encode("utf-8")).decode()
+    if shell_family(kind) == "bash":
+        return f'eval "$(printf %s {encoded} | base64 -d)"'
+    return (". ([scriptblock]::Create([Text.Encoding]::UTF8.GetString("
+            f"[Convert]::FromBase64String('{encoded}'))))")
+
+
+VISIBLE = re.compile(rb"\x1b\][^\x07\x1b]*(?:\x07|\x1b\\\\)|\x1b\[[0-?]*[ -/]*[@-~]|\r")
+
+
+def at_prompt(log):
+    """True when the visible text so far ends in a shell prompt (`$` or `>`)."""
+    return VISIBLE.sub(b"", log).rstrip(b" \t\n").endswith((b"$", b">"))
+
+
+def prompts(log):
+    """Every status the installed prompt has reported in these bytes, oldest first."""
+    return [dict(n=int(m[1]), ok=m[2] == b"1", exit_code=int(m[3]) if m[3] else None,
+                 cwd=m[4].decode("utf-8", "replace")) for m in MARKER.finditer(log)]
+
+
+def key_params(key):
+    if key in KEYS:
+        return dict(KEYS[key])
+    if len(key) == 1 and key.isprintable():
+        return dict(key=key, text=key)
+    raise SystemExit(f"unknown key {key!r}: use one character or one of {', '.join(KEYS)}")
+
+
+def film(out, seconds, hold, capture, finished, clock=time.monotonic, sleep=time.sleep):
+    """Write PNG frames on the FPS grid until `finished()` plus `hold` seconds,
+    or `seconds` in all. A slow capture repeats the previous frame, so the
+    directory plays back at exactly FPS. Returns the frame count."""
+    out.mkdir(parents=True, exist_ok=True)
+    start, index, last, stop = clock(), -1, None, None
+    while True:
+        now = clock()
+        if stop is None and finished():
+            stop = now + hold
+        if now - start >= seconds or (stop is not None and now >= stop):
+            return index + 1
+        slot = int((now - start) * FPS)
+        if slot > index:
+            png = capture()
+            for missed in range(index + 1, slot):
+                (out / f"f{missed:05d}.png").write_bytes(last or png)
+            (out / f"f{slot:05d}.png").write_bytes(png)
+            index, last = slot, png
+        sleep(0.02)
 
-def http_json(url):
-    with urllib.request.urlopen(url, timeout=1) as response:
-        return json.load(response)
 
 class CDP:
-    def __init__(self, url, on_event, trace_path):
+    def __init__(self, url):
         import websocket
 
         self.ws = websocket.create_connection(url, timeout=5, suppress_origin=True)
-        self.ws.settimeout(0.02)
-        self.on_event, self.counter, self.responses = on_event, 0, {}
-        self.trace = trace_path.open("w", encoding="utf-8")
-
-    def log(self, direction, **fields):
-        self.trace.write(json.dumps({"time": time.time(), "direction": direction, **fields}) + "\n")
-        self.trace.flush()
-
-    def send(self, method, params=None):
-        self.counter += 1
-        self.ws.send(json.dumps({"id": self.counter, "method": method, "params": params or {}}))
-        self.log("sent", id=self.counter, method=method,
-                 frame_session_id=(params or {}).get("sessionId"))
-        return self.counter
-
-    def pump(self):
+        self.count, self.on_event = 0, None
+
+    def recv(self, timeout):
         import websocket
 
+        self.ws.settimeout(timeout)
         try:
             raw = self.ws.recv()
         except websocket.WebSocketTimeoutException:
-            self.log("receive_timeout")
-            return
+            return None
         if not raw:
-            raise ConnectionError("Browser CDP connection closed")
-        event = json.loads(raw)
-        self.log("received", id=event.get("id"), method=event.get("method"),
-                 frame_session_id=event.get("params", {}).get("sessionId"))
-        if "id" in event:
-            self.responses[event["id"]] = event
-        else:
-            self.on_event(event)
-
-    def call(self, method, params=None, timeout=5):
-        request = self.send(method, params)
+            raise ConnectionError("browser connection closed")
+        message = json.loads(raw)
+        if "id" not in message and self.on_event:
+            self.on_event(message)
+        return message
+
+    def call(self, method, params=None, timeout=10):
+        self.count += 1
+        self.ws.send(json.dumps({"id": self.count, "method": method, "params": params or {}}))
         deadline = time.monotonic() + timeout
-        while request not in self.responses and time.monotonic() < deadline:
-            self.pump()
-        response = self.responses.pop(request, None)
-        if response is None or "error" in response:
-            raise RuntimeError(f"CDP {method}: {response}")
-        return response.get("result", {})
-
-class Terminal:
-    def __init__(self, args, directory):
-        self.directory, self.args = directory, args
-        directory.mkdir(parents=True, exist_ok=False)
-        self.session = uuid.uuid4().hex
-        self.parser = CompletionParser()
-        self.request_id = None
-        self.socket_ids = []
-        self.closed = False
-        self.cdp = None
-        self.raw = (directory / "network.jsonl").open("w", encoding="utf-8")
-        self.output = (directory / "terminal.bin").open("wb")
-        self.job = WindowsJob()
-        self.launches = []
-        self.take = None
-        self.terminal_sizes = []
-
-    def event(self, event):
-        method, params = event["method"], event.get("params", {})
-        if method.startswith("Network.webSocket"):
-            self.raw.write(json.dumps(event, ensure_ascii=False) + "\n")
-            self.raw.flush()
-        if method == "Network.webSocketCreated" and params["url"] == self.terminal_url.replace("http:", "ws:") + "ws":
-            self.socket_ids.append(params["requestId"])
-            if self.request_id is None:
-                self.request_id = params["requestId"]
-            else:
-                self.closed = True  # A reconnect is never continuity.
-        if params.get("requestId") != self.request_id or self.request_id is None:
-            return
-        if method == "Network.webSocketClosed":
-            self.closed = True
-        if method == "Network.webSocketFrameSent":
-            frame = params["response"]
-            raw = base64.b64decode(frame["payloadData"]) if frame["opcode"] == 2 else frame["payloadData"].encode("utf-8")
-            payload = raw[1:] if raw[:1] == b"1" else raw
-            if payload[:1] == b"{":
-                size = json.loads(payload)
-                if "columns" in size and "rows" in size:
-                    self.terminal_sizes.append({"columns": size["columns"], "rows": size["rows"],
-                                                "observed_at": time.time(), "request_id": self.request_id})
-        if method == "Network.webSocketFrameReceived":
-            frame = params["response"]
-            raw = base64.b64decode(frame["payloadData"]) if frame["opcode"] == 2 else frame["payloadData"].encode("utf-8")
-            data = ttyd_output(event)
-            self.output.write(data)
-            self.output.flush()
-            self.parser.feed(data)
-
-    def start(self):
-        port, debug_port = free_port(), free_port()
-        self.terminal_url = f"http://127.0.0.1:{port}/"
-        shell_argv = [str(self.args.shell_exe)] + (["--noprofile", "--norc", "-i"] if self.args.shell_kind == "gitbash" else ["-NoLogo", "-NoProfile", "-NoExit"])
-        # ttyd 1.7.7 needs -w but decodes its argv using the ANSI code page.
-        # Inherit the Unicode cwd through CreateProcessW, then use a relative -w.
-        ttyd_argv = [str(self.args.ttyd), "-i", "127.0.0.1", "-p", str(port), "-W", "-m", "1",
-                     "-w", "."] + shell_argv
-        browser_argv = [str(self.args.browser), "--headless=new", "--no-first-run", "--no-default-browser-check",
-                        "--disable-background-networking", "--remote-debugging-address=127.0.0.1",
-                        f"--remote-debugging-port={debug_port}", f"--user-data-dir={self.directory / 'profile'}",
-                        "--window-size=1600,900", "about:blank"]
-        for name, argv in (("ttyd", ttyd_argv), ("browser", browser_argv)):
-            cwd = self.args.cwd if name == "ttyd" else self.directory
-            pid = self.job.spawn(argv, cwd, self.directory / f"{name}.log")
-            self.launches.append({"name": name, "argv": argv, "pid": pid})
-        write_json(self.directory / "launches.json", self.launches)
-        deadline = time.monotonic() + 20
-        last_error = None
         while time.monotonic() < deadline:
+            message = self.recv(0.05)
+            if message and message.get("id") == self.count:
+                if "error" in message:
+                    raise RuntimeError(f"{method}: {message['error']}")
+                return message.get("result", {})
+        raise TimeoutError(f"{method} took longer than {timeout:g}s")
+
+
+def free_port():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+def page_url(debug_port):
+    with urllib.request.urlopen(f"http://127.0.0.1:{debug_port}/json/list", timeout=2) as response:
+        pages = json.load(response)
+    return next(page["webSocketDebuggerUrl"] for page in pages if page["type"] == "page")
+
+
+def connect(session):
+    return CDP(page_url(session["debug_port"]))
+
+
+def type_text(cdp, text):
+    cdp.call("Runtime.evaluate", {"expression": "document.querySelector('.xterm-helper-textarea').focus()"})
+    cdp.call("Input.insertText", {"text": text})
+
+
+def press(cdp, key):
+    params = key_params(key)
+    cdp.call("Input.dispatchKeyEvent", dict(type="keyDown", **params))
+    cdp.call("Input.dispatchKeyEvent", dict(type="keyUp", **{k: v for k, v in params.items() if k != "text"}))
+
+
+def screenshot(cdp):
+    return base64.b64decode(cdp.call("Page.captureScreenshot", {"format": "png"}, timeout=5)["data"])
+
+
+def tail(path, size=262144):
+    with path.open("rb") as handle:
+        handle.seek(max(0, handle.seek(0, os.SEEK_END) - size))
+        return handle.read()
+
+
+def read_json(path):
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def write_json(path, value):
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def load_session(directory):
+    if not (directory / "ready.json").is_file():
+        raise SystemExit(f"{directory} has no ready.json: is `serve` running there?")
+    return read_json(directory / "session.json")
+
+
+def lit_fraction(png):
+    from PIL import Image
+
+    image = Image.open(io.BytesIO(png)).convert("L")
+    return sum(image.histogram()[91:]) / (image.width * image.height)
+
+
+def serve(args):
+    directory = args.session
+    if directory.exists() and any(directory.iterdir()):
+        raise SystemExit(f"{directory} is not empty: use a new session directory")
+    directory.mkdir(parents=True, exist_ok=True)
+    cwd = Path(args.cwd or os.getcwd()).resolve()
+    if not cwd.is_dir():
+        raise SystemExit(f"--cwd is not a directory: {cwd}")
+    shell = shell_argv(args.shell, args.shell_exe)
+    ttyd = args.ttyd or shutil.which("ttyd")
+    if not ttyd:
+        raise SystemExit("ttyd is not on PATH; pass --ttyd")
+    browser = find_browser(args.browser)
+    if not browser:
+        raise SystemExit("no Chrome or Edge found; pass --browser")
+    port, debug_port = free_port(), free_port()
+    unix = os.name != "nt"
+    # ttyd 1.7 on Windows needs -w but decodes it in the ANSI code page, so
+    # pass "." and let the shell inherit this process's Unicode cwd; the
+    # installed prompt script then cds there explicitly and reports back.
+    ttyd_argv = [ttyd, "-i", "127.0.0.1", "-p", str(port), "-W", "-m", "1", "-w", ".",
+                 "-t", "fontSize=17", *shell]
+    # Software GL: without it a GPU-less session paints the xterm canvas empty.
+    browser_argv = [browser, "--headless=new", "--no-first-run", "--no-default-browser-check",
+                    "--use-gl=angle", "--use-angle=swiftshader", "--enable-unsafe-swiftshader",
+                    "--disable-background-networking", "--remote-debugging-address=127.0.0.1",
+                    f"--remote-debugging-port={debug_port}", f"--user-data-dir={directory / 'profile'}",
+                    f"--window-size={WIDTH},{HEIGHT}", "--hide-scrollbars", "about:blank"]
+    logs = [(directory / "ttyd.log").open("ab"), (directory / "browser.log").open("ab")]
+    processes = [
+        subprocess.Popen(ttyd_argv, cwd=cwd, stdin=subprocess.DEVNULL, stdout=logs[0],
+                         stderr=subprocess.STDOUT, start_new_session=unix),
+        subprocess.Popen(browser_argv, cwd=directory, stdin=subprocess.DEVNULL, stdout=logs[1],
+                         stderr=subprocess.STDOUT, start_new_session=unix),
+    ]
+    session = dict(shell=args.shell, cwd=str(cwd), terminal_url=f"http://127.0.0.1:{port}/",
+                   debug_port=debug_port, pids=[process.pid for process in processes])
+    write_json(directory / "session.json", session)
+    output = (directory / "terminal.log").open("ab")
+    state = {"closed": False}
+
+    def on_event(event):
+        if event["method"] == "Network.webSocketFrameReceived":
+            frame = event["params"]["response"]
+            raw = base64.b64decode(frame["payloadData"]) if frame["opcode"] == 2 else frame["payloadData"].encode()
+            if raw[:1] == b"0":  # ttyd frame type 0 is terminal output
+                output.write(raw[1:])
+                output.flush()
+        elif event["method"] == "Network.webSocketClosed":
+            state["closed"] = True
+
+    def pump_until(condition, timeout, failure):
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            cdp.recv(0.05)
+            if state["closed"]:
+                raise ConnectionError("the terminal connection closed")
+            if condition():
+                return
+        (directory / "timeout.png").write_bytes(screenshot(cdp))
+        raise TimeoutError(f"{failure}; terminal output so far: {tail(directory / 'terminal.log')[-300:]!r}")
+
+    def pump_while_output_flows(quiet):
+        # A shell's banner and first prompt can trickle out; type only once
+        # it has been silent for `quiet` seconds.
+        deadline, seen = time.monotonic() + quiet, output.tell()
+        while time.monotonic() < deadline:
+            cdp.recv(0.05)
+            if state["closed"]:
+                raise ConnectionError("the terminal connection closed")
+            if output.tell() != seen:
+                deadline, seen = time.monotonic() + quiet, output.tell()
+
+    code = 1
+    try:
+        deadline = time.monotonic() + 20
+        while True:
             try:
-                pages = http_json(f"http://127.0.0.1:{debug_port}/json/list")
-                page = next(p for p in pages if p["type"] == "page")
-                self.cdp = CDP(page["webSocketDebuggerUrl"], self.event, self.directory / "cdp-trace.jsonl")
+                cdp = CDP(page_url(debug_port))
                 break
             except (OSError, StopIteration) as error:
-                last_error = error
+                if time.monotonic() > deadline:
+                    raise TimeoutError(f"browser did not start: {error}") from None
                 time.sleep(0.1)
-        if self.cdp is None:
-            raise TimeoutError(f"Browser startup: {last_error}")
-        self.page_id = page["id"]
-        self.cdp.call("Network.enable")  # Must precede navigation and terminal socket creation.
-        self.cdp.call("Page.enable")
-        self.cdp.call("Emulation.setDeviceMetricsOverride", {"width": 1600, "height": 900, "deviceScaleFactor": 1, "mobile": False})
-        self.cdp.call("Page.navigate", {"url": self.terminal_url})
-        deadline = time.monotonic() + 10
-        while time.monotonic() < deadline:
-            self.cdp.pump()
-            if self.parser.text or self.closed:
-                break
-        self.screenshot("startup.png")
-        if self.closed:
-            raise ConnectionError("Terminal WebSocket closed before readiness")
-        if not self.parser.text:
-            raise TimeoutError("No ttyd output from the filmed terminal within 10 seconds")
-
-    def screenshot(self, name):
-        result = self.cdp.call("Page.captureScreenshot", {"format": "png"})
-        (self.directory / name).write_bytes(base64.b64decode(result["data"]))
-
-    def type(self, text):
-        self.cdp.call("Runtime.evaluate", {"expression": "document.querySelector('.xterm-helper-textarea').focus()"})
-        self.cdp.call("Input.insertText", {"text": text})
-        self.cdp.call("Input.dispatchKeyEvent", {"type": "keyDown", "key": "Enter", "code": "Enter", "windowsVirtualKeyCode": 13, "text": "\r"})
-        self.cdp.call("Input.dispatchKeyEvent", {"type": "keyUp", "key": "Enter", "code": "Enter", "windowsVirtualKeyCode": 13})
-
-    def readiness(self):
-        if self.args.shell_kind == "gitbash":
-            # Native Windows Python is explicit even inside Git Bash.
-            python = str(Path(sys.executable)).replace("\\", "/")
-            code = "import base64,json,os;print('[MOVIE|'+base64.b64encode(json.dumps(dict(session=os.environ['MOVIE_SESSION'],pid=os.environ['MOVIE_SHELL_PID'],shell='gitbash',cwd=os.getcwd())).encode()).decode()+'|END]')"
-            # Encode the Python payload to keep the complete framing out of input echo.
-            payload = base64.b64encode(code.encode()).decode()
-            command = "cd -- " + shlex.quote(str(self.args.cwd).replace("\\", "/")) + " && " + f"export MOVIE_SESSION={self.session} MOVIE_SHELL_PID=$$; '{python}' -c \"import base64;exec(base64.b64decode('{payload}'))\""
-        else:
-            script = "Set-Location -LiteralPath '" + str(self.args.cwd).replace("'", "''") + "' -ErrorAction Stop; $global:MovieSession='" + self.session + "'; $r=@{session=$MovieSession;pid=$PID;shell=$PSVersionTable.PSVersion.ToString();cwd=(Get-Location).Path}; [Console]::WriteLine('[MOVIE|'+[Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes(($r|ConvertTo-Json -Compress)))+'|END]')"
-            payload = base64.b64encode(script.encode("utf-8")).decode()
-            command = ". ([scriptblock]::Create([Text.Encoding]::UTF8.GetString([Convert]::FromBase64String('" + payload + "'))))"
-        write_json(self.directory / "readiness-command.json", {"command": command, "session": self.session})
-        self.type(command)
-        deadline = time.monotonic() + 10
-        while time.monotonic() < deadline:
-            self.cdp.pump()
-            matches = [r for r in self.parser.records if r.get("session") == self.session]
-            if matches:
-                if os.path.normcase(os.path.abspath(matches[-1].get("cwd", ""))) != os.path.normcase(str(self.args.cwd.resolve())):
-                    raise RuntimeError("Recorded shell did not enter the requested cwd")
-                self.screenshot("ready.png")
-                return matches[-1]
-            if self.closed:
-                raise ConnectionError("Terminal connection lost during readiness")
-        self.screenshot("readiness-timeout.png")
-        raise TimeoutError("No nonce/shell/cwd completion record from the filmed terminal")
-
-    def wait_record(self, predicate, timeout=10):
-        deadline = time.monotonic() + timeout
-        while time.monotonic() < deadline:
-            matches = [record for record in self.parser.records if record.get("session") == self.session and predicate(record)]
-            if matches:
-                return matches[-1]
-            if self.closed:
-                raise ConnectionError("Terminal connection lost while waiting for completion")
-            self.cdp.pump()
-        raise TimeoutError("Missing terminal completion record")
-
-    def install_prompt(self):
-        if self.args.shell_kind == "gitbash":
-            python = str(Path(sys.executable)).replace("\\", "/")
-            source = str(Path(__file__).resolve()).replace("\\", "/")
-            script = f'''export MOVIE_SESSION={self.session}
-MOVIE_PHASE=boot
-movie_prompt() {{
-    local movie_status=$? movie_pipeline=("${{PIPESTATUS[@]}}")
-    if [[ -n "$MOVIE_PHASE" ]]; then
-        MOVIE_STATUS="$movie_status" MOVIE_PIPELINE="${{movie_pipeline[*]}}" MOVIE_SHELL_PID=$$ MOVIE_CWD="$PWD" MOVIE_PHASE="$MOVIE_PHASE" MOVIE_REQUEST="$MOVIE_REQUEST" MOVIE_NATIVE="$MOVIE_NATIVE" '{python}' '{source}' --emit-bash
-        if [[ "$MOVIE_PHASE" == arm ]]; then MOVIE_PHASE=run; else MOVIE_PHASE=; fi
-    fi
-    return "$movie_status"
-}}
-PROMPT_COMMAND=movie_prompt
-PS1='MOVIE $ '
-'''
-            payload = base64.b64encode(script.encode()).decode()
-            self.type(f'''eval "$('{python}' -c "import base64;print(base64.b64decode('{payload}').decode())")"''')
-        else:
-            script = r'''
-$global:MovieErrorCount=$Error.Count
-$global:MoviePhase='boot'
-function global:prompt {
-    $movieOK=$?
-    $movieNative=$global:LASTEXITCODE
-    if ($global:MoviePhase) {
-        $movieError=$null
-        $parseError=$false
-        if ($Error.Count -gt 0 -and $Error.Count -gt $global:MovieErrorCount) {
-            $movieError=$Error[0].ToString()
-            $parseError=($Error[0] -is [System.Management.Automation.ParseException] -or $Error[0].Exception -is [System.Management.Automation.ParseException] -or $Error[0].CategoryInfo.Category -eq 'ParserError')
-        }
-        $record=@{session=$global:MovieSession;request=$global:MovieRequest;phase=$global:MoviePhase;pid=$PID;cwd=(Get-Location).Path;outcome='completed';shell_success=($movieOK -and -not $parseError);raw_shell_success=$movieOK;parse_error=$parseError;native_exit_code=$null;shell_error=$movieError;producer_exit_code=$null;producer_success=$null}
-        if ($global:MovieNative -and -not $parseError) {$record.native_exit_code=$movieNative;$record.producer_exit_code=$movieNative;$record.producer_success=($movieNative -eq 0)}
-        if ($global:MoviePhase -eq 'arm') {$global:MoviePhase='run'} else {$global:MoviePhase=$null}
-        $encoded=[Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes(($record|ConvertTo-Json -Compress)))
-        [Console]::WriteLine('[MOVIE|')
-        for ($offset=0;$offset -lt $encoded.Length;$offset+=60) {[Console]::WriteLine($encoded.Substring($offset,[Math]::Min(60,$encoded.Length-$offset)))}
-        [Console]::WriteLine('|END]')
-    }
-    return 'MOVIE PS> '
-}
-'''
-            payload = base64.b64encode(script.encode()).decode()
-            self.type(". ([scriptblock]::Create([Text.Encoding]::UTF8.GetString([Convert]::FromBase64String('" + payload + "'))))")
-        try:
-            return self.wait_record(lambda r: r.get("phase") == "boot")
-        except TimeoutError:
-            if self.args.shell_kind != "gitbash":
-                self.type("$Error | Select-Object -First 3 | Format-List * -Force")
-                deadline = time.monotonic() + 2
-                while time.monotonic() < deadline:
-                    self.cdp.pump()
-                self.screenshot("prompt-error.png")
-            raise
-
-    def arm(self, request, native):
-        if self.args.shell_kind == "gitbash":
-            command = f"MOVIE_REQUEST={request}; MOVIE_NATIVE={int(bool(native))}; MOVIE_PHASE=arm"
-        else:
-            reset_native = "$global:LASTEXITCODE=$null; " if native else ""
-            command = reset_native + f"$global:MovieRequest={request}; $global:MovieNative=${str(bool(native)).lower()}; $global:MovieErrorCount=$Error.Count; $global:MoviePhase='arm'"
-        self.type(command)
-        self.wait_record(lambda r: r.get("phase") == "arm" and str(r.get("request")) == str(request))
-
-    def close(self):
-        start = time.monotonic()
-        try:
-            self.job.close()
-        finally:
-            if self.cdp is not None:
-                self.cdp.ws.close()
-                self.cdp.trace.close()
-            self.output.close()
-            self.raw.close()
-        return time.monotonic() - start
-
-def emit_bash():
-    env = os.environ
-    status = int(env["MOVIE_STATUS"])
-    native = env.get("MOVIE_NATIVE") == "1"
-    pipeline = [int(value) for value in env["MOVIE_PIPELINE"].split()]
-    record = dict(session=env["MOVIE_SESSION"], request=env.get("MOVIE_REQUEST"),
-                  phase=env["MOVIE_PHASE"], pid=env["MOVIE_SHELL_PID"],
-                  pid_kind="msys", cwd=env["MOVIE_CWD"],
-                  outcome="completed", shell_success=status == 0,
-                  native_exit_code=pipeline[0] if native else None, shell_error=None,
-                  pipeline_statuses=pipeline, producer_exit_code=pipeline[0] if native else None,
-                  producer_success=pipeline[0] == 0 if native else None)
-    encoded = base64.b64encode(json.dumps(record).encode()).decode()
-    print("[MOVIE|\n" + "\n".join(encoded[offset:offset + 60] for offset in range(0, len(encoded), 60)) + "\n|END]", flush=True)
-
-
-class Recorder:
-    def __init__(self, args):
-        self.args = args
-        self.terminal = Terminal(args, args.directory)
-        self.control = args.directory / 'control'
-        self.control.mkdir()
-        self.next_id = 1
-        self.pending = None
-        self.take = None
-        self.capture = None
-        self.capture_due = 0
-        self.stopping = None
-        self.geometry = None
-        self.readiness = None
-        self.closed = False
-
-    def reply(self, number, value, suffix='result'):
-        value = dict(value, id=number, session_id=self.terminal.session)
-        write_json(self.control / f'{number:06d}.{suffix}.json', value)
-        return value
-
-    def status(self):
-        value = dict(session_id=self.terminal.session, next_request_id=self.next_id,
-                     pending=self.pending, active_take=self.take and self.take['name'],
-                     geometry=self.geometry, readiness=self.readiness,
-                     pid=os.getpid(), closed=self.closed)
-        write_json(self.control / 'status.json', value)
-        write_json(self.control / 'ready.json', value)
-        return value
-
-    def begin(self, name, number):
-        if self.take is not None:
-            raise ValueError('A take is already active')
-        if not isinstance(name, str) or not re.fullmatch(r'[a-zA-Z0-9_-]+', name):
-            raise ValueError('Take name must contain only letters, digits, underscore, or dash')
-        directory = self.args.directory / name
-        directory.mkdir()
-        (directory / 'samples').mkdir()
-        self.take = dict(name=name, directory=str(directory), samples=[], begin_request=number)
-        self.capture_due = 0
-
-    def capture_tick(self, schedule=True):
-        if self.take is None:
-            return
-        now = time.monotonic()
-        if self.capture is not None:
-            number, requested = self.capture
-            if now - requested > 2:
-                raise TimeoutError('Screenshot response exceeded two seconds')
-            response = self.terminal.cdp.responses.pop(number, None)
-            if response is not None:
-                if 'error' in response:
-                    raise RuntimeError(f'Screenshot failed: {response["error"]}')
-                samples = self.take['samples']
-                path = Path(self.take['directory']) / 'samples' / f'{len(samples):06d}.png'
-                path.write_bytes(base64.b64decode(response['result']['data']))
-                samples.append(dict(requested=requested, completed=now, path=str(path)))
-                write_json(Path(self.take['directory']) / 'capture.json', self.take)
-                self.capture = None
-                if len(samples) == 1:
-                    self.take['start'] = now
-                    self.status()
-                    self.reply(self.take['begin_request'], dict(outcome='completed', success=True,
-                               start=now, name=self.take['name']))
-        if schedule and self.capture is None and now >= self.capture_due:
-            self.capture = (self.terminal.cdp.send('Page.captureScreenshot', {'format': 'png'}), now)
-            self.capture_due = now + .2
-
-    def check_observation(self):
-        if self.terminal.closed:
-            raise ConnectionError('Terminal connection lost or reconnected')
-        if any({k:size[k] for k in self.geometry} != self.geometry
-               for size in self.terminal.terminal_sizes):
-            raise RuntimeError('Terminal geometry changed; start a new fixed-viewport session')
-
-    def end(self, incomplete=False):
-        if self.take is None:
-            raise ValueError('No take is active')
-        if not incomplete:
-            while self.capture is not None:
-                self.terminal.cdp.pump()
-                self.check_observation()
-                self.capture_tick(schedule=False)
-            self.check_observation()
-        take = self.take
-        end = time.monotonic()
-        take['end'] = end
-        take['incomplete'] = incomplete
-        directory = Path(take['directory'])
-        try:
-            if not incomplete:
-                completed = [sample['completed'] for sample in take['samples']]
-                sources = frame_sources(completed, take.get('start'), end)
-                for index, source in enumerate(sources):
-                    shutil.copyfile(take['samples'][source]['path'], directory / f'{index:06d}.png')
-                take.update(frame_sources=sources, duplicated_intervals=[i for i in range(1,len(sources)) if sources[i] == sources[i-1]],
-                            duration=end-take['start'], kind='frames', src=str(directory), rate=5)
-            write_json(directory / 'take.json', take)
-        except BaseException:
-            take['incomplete'] = True
-            raise
-        finally:
-            self.take = None
-            self.capture = None
-        return dict(take, outcome='interrupted' if incomplete else 'completed', success=not incomplete)
-
-    def finish_pending(self):
-        if self.pending is None:
-            return
-        pending = self.pending
-        records = [r for r in self.terminal.parser.records
-                   if r.get('session') == self.terminal.session and r.get('phase') == 'run'
-                   and str(r.get('request')) == str(pending['id'])]
-        if records:
-            result = dict(records[-1], native_producer=pending.get('native_producer'), command=pending['command'])
-            result['success'] = command_succeeded(result)
-            self.pending = None
-            self.status()
-            self.reply(pending['id'], result)
-        elif time.monotonic() >= pending['deadline']:
-            self.interrupt_pending('unknown', 'Command completion deadline expired')
-            raise TimeoutError('Command completion deadline expired; session terminated')
-
-    def interrupt_pending(self, outcome, reason):
-        if self.pending:
-            pending = self.pending
-            self.pending = None
-            self.reply(pending['id'], dict(outcome=outcome, shell_success=None, raw_shell_success=None,
-                       shell_error=None, native_producer=pending.get('native_producer'),
-                       producer_exit_code=None, success=False, reason=reason))
-
-    def dispatch(self, request):
-        number, operation = request['id'], request['operation']
-        if operation == 'run':
-            if self.pending is not None:
-                raise ValueError('Another command still owns terminal input')
-            command = request.get('command')
-            timeout = request.get('timeout_seconds', 90)
-            if not isinstance(command, str) or not command or not isinstance(timeout, (int,float)) or not math.isfinite(timeout) or timeout <= 0:
-                raise ValueError('run requires a command and positive finite timeout_seconds')
-            native = request.get('native_producer')
-            if native is not None and (not isinstance(native,str) or not native):
-                raise ValueError('native_producer must name the direct or first pipeline executable')
-            self.terminal.arm(number, native)
-            self.pending = dict(request, deadline=time.monotonic()+timeout)
-            self.terminal.type(command)
-        elif operation == 'begin-take':
-            self.begin(request.get('name'), number)
-        elif operation == 'end-take':
-            if self.take is None or not self.take['samples']:
-                raise ValueError('No started take is active')
-        elif operation == 'key':
-            validate_key(request.get('key'))
-        elif operation not in ('inspect', 'close', 'cancel'):
-            raise ValueError(f'Unknown operation: {operation}')
-        self.status()
-        self.reply(number, dict(accepted=True, operation=operation), 'ack')
-        if operation in ('run', 'begin-take'):
-            return
-        if operation == 'end-take':
-            value = self.end()
-        elif operation == 'key':
-            send_key(self.terminal, request['key'])
-            value = dict(outcome='completed', success=True)
-        elif operation == 'inspect':
-            value = dict(self.status(), outcome='completed', success=True)
-        else:
-            self.stopping = request
-            return
-        self.status()
-        self.reply(number, value)
-
-    def read_request(self):
-        path = self.control / f'{self.next_id:06d}.request.json'
-        if not path.exists():
-            return
-        request = read_json(path)
-        number = self.next_id
-        self.next_id += 1
-        try:
-            if request.get('id') != number:
-                raise ValueError('Request ID does not match ordered request file')
-            self.dispatch(request)
-        except (ValueError, FileExistsError) as error:
-            self.status()
-            self.reply(number, dict(accepted=False, reason=str(error)), 'ack')
-            self.reply(number, dict(outcome='completed', success=False, reason=str(error)))
-
-    def run(self):
-        failure = None
-        try:
-            self.terminal.start()
-            self.readiness = self.terminal.readiness()
-            self.terminal.install_prompt()
-            sizes = self.terminal.terminal_sizes
-            if not sizes or sizes[-1]['columns'] < 80:
-                raise RuntimeError('Terminal requires at least 80 columns at the fixed 1600x900 viewport')
-            self.geometry = {k:sizes[-1][k] for k in ('columns','rows')}
-            self.status()
-            while self.stopping is None:
-                self.terminal.cdp.pump()
-                self.check_observation()
-                self.finish_pending()
-                self.capture_tick()
-                self.read_request()
-            self.interrupt_pending('interrupted', 'Session closed by controller')
-            if self.take:
-                self.end(incomplete=self.stopping['operation']=='cancel')
-        except BaseException as error:
-            failure = error
+        cdp.on_event = on_event
+        cdp.call("Network.enable")  # before navigation, or the terminal socket is never reported
+        cdp.call("Page.enable")
+        cdp.call("Emulation.setDeviceMetricsOverride",
+                 {"width": WIDTH, "height": HEIGHT, "deviceScaleFactor": 1, "mobile": False})
+        cdp.call("Page.navigate", {"url": session["terminal_url"]})
+        # Type only once the shell is reading input: its own prompt is on
+        # screen and nothing more has arrived for a moment.
+        pump_until(lambda: at_prompt(tail(directory / "terminal.log")), 20, "the shell never showed a prompt")
+        pump_while_output_flows(0.5)
+        # Git Bash under ConPTY can lose the first keystroke of a session.
+        # Spend it on a bare Enter, which only repaints the prompt.
+        press(cdp, "Enter")
+        pump_while_output_flows(0.5)
+        type_text(cdp, prompt_command(args.shell, cwd))
+        press(cdp, "Enter")
+        pump_until(lambda: prompts(tail(directory / "terminal.log")), 15,
+                   "the shell never showed the installed prompt")
+        prompt = prompts(tail(directory / "terminal.log"))[-1]
+        if shell_family(args.shell) == "powershell":
+            entered = os.path.normcase(os.path.normpath(prompt["cwd"])) == os.path.normcase(str(cwd))
+        else:  # Git Bash reports /c/... paths, so compare the leaf directory
+            entered = Path(prompt["cwd"]).name == cwd.name
+        if not entered:
+            raise RuntimeError(f"the shell is in {prompt['cwd']!r}, not {str(cwd)!r}")
+
+        def typed(text):  # type one line and wait for the prompt after it
+            before = prompts(tail(directory / "terminal.log"))[-1]["n"]
+            type_text(cdp, text)
+            press(cdp, "Enter")
+            pump_until(lambda: prompts(tail(directory / "terminal.log"))[-1]["n"] > before, 15,
+                       f"no prompt after typing {text[:24]!r}")
+
+        # Preflight as the Unix route does: print something dense, refuse a blank canvas.
+        typed("echo '" + "#" * 120 + "'")
+        time.sleep(0.3)
+        png = screenshot(cdp)
+        (directory / "ready.png").write_bytes(png)
+        lit = lit_fraction(png)
+        if lit < 0.002:
+            raise RuntimeError(f"the terminal renders blank ({lit:.4%} lit pixels); see ready.png")
+        typed("clear")
+        time.sleep(0.3)
+        prompt = prompts(tail(directory / "terminal.log"))[-1]
+        write_json(directory / "ready.json", dict(session, prompt=prompt, lit=round(lit, 4)))
+        print(json.dumps({"ready": True, "session": str(directory), "cwd": prompt["cwd"]},
+                         ensure_ascii=False), flush=True)
+        while not (directory / "stop").exists():
+            cdp.recv(0.2)
+            if state["closed"]:
+                raise ConnectionError("the terminal connection closed")
+        code = 0
+    except KeyboardInterrupt:
+        code = 0
+    except Exception as error:  # noqa: BLE001 - report, then clean up below
+        print(f"serve: {error}", file=sys.stderr)
+        code = 0 if (directory / "stop").exists() else 1
+    finally:
+        for process in processes:
+            kill_process_tree(process.pid)
             try:
-                self.interrupt_pending('unknown', str(error))
-            except BaseException:
+                process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
                 pass
-            if self.take:
-                try:
-                    self.end(incomplete=True)
-                except BaseException:
-                    pass
-        finally:
-            try:
-                self.terminal.close()
-            except BaseException as error:
-                failure = failure or error
-            self.closed = True
-            try:
-                self.status()
-                if self.stopping:
-                    self.reply(self.stopping['id'], dict(outcome='completed', success=failure is None,
-                               reason=str(failure) if failure else None))
-            except BaseException as error:
-                failure = failure or error
-        if failure:
-            raise failure
-
-
-SPECIAL_KEYS = {
-    'Enter': dict(key='Enter', code='Enter', windowsVirtualKeyCode=13, text='\r'),
-    'Escape': dict(key='Escape', code='Escape', windowsVirtualKeyCode=27),
-    'Tab': dict(key='Tab', code='Tab', windowsVirtualKeyCode=9, text='\t'),
-    'ArrowLeft': dict(key='ArrowLeft', code='ArrowLeft', windowsVirtualKeyCode=37),
-    'ArrowUp': dict(key='ArrowUp', code='ArrowUp', windowsVirtualKeyCode=38),
-    'ArrowRight': dict(key='ArrowRight', code='ArrowRight', windowsVirtualKeyCode=39),
-    'ArrowDown': dict(key='ArrowDown', code='ArrowDown', windowsVirtualKeyCode=40),
-    'Ctrl-C': dict(key='c', code='KeyC', windowsVirtualKeyCode=67, modifiers=2),
-}
-
-
-def validate_key(key):
-    if not isinstance(key, str) or not (key in SPECIAL_KEYS or len(key)==1 and key.isprintable()):
-        raise ValueError('key must be printable, Enter, Escape, Tab, ArrowLeft/Up/Right/Down, or Ctrl-C')
+        for handle in (output, *logs):
+            handle.close()
+    return code
 
 
-def send_key(terminal, key):
-    validate_key(key)
-    params = SPECIAL_KEYS.get(key, dict(key=key, text=key))
-    terminal.cdp.call('Input.dispatchKeyEvent', dict(type='keyDown', **params))
-    terminal.cdp.call('Input.dispatchKeyEvent', dict(type='keyUp', **{k:v for k,v in params.items() if k!='text'}))
+def observe(args, cdp, n0):
+    """Film or wait until the prompt after `n0` appears; print the status."""
+    log = args.session / "terminal.log"
 
+    def latest():
+        return next((p for p in reversed(prompts(tail(log))) if p["n"] > n0), None)
 
-def wait_reply(control, number, suffix, timeout):
-    deadline = time.monotonic() + timeout
-    path = control / f'{number:06d}.{suffix}.json'
-    while True:
-        if path.exists():
-            return read_json(path)
-        if time.monotonic() >= deadline:
-            return dict(id=number, outcome='client-timeout', success=False,
-                        reason='Client wait expired; the command was not cancelled. Use result --id to wait again.')
-        time.sleep(.02)
-
-
-def client(args):
-    deadline = time.monotonic() + args.timeout
-    control = args.directory / 'control'
-    if args.action == 'request':
-        request = read_json(args.file)
-        status = read_json(control / 'status.json')
-        number = request.get('id')
-        if type(number) is not int or number != status['next_request_id'] or status.get('closed'):
-            raise ValueError(f"Request ID must be next_request_id={status['next_request_id']} in an open session")
-        path = control / f'{number:06d}.request.json'
-        if path.exists():
-            raise ValueError('Request ID already submitted; use result --id')
-        write_json(path, request)
-        reply = wait_reply(control, number, 'ack', max(0, deadline-time.monotonic()))
-        if reply.get('accepted') is not True:
-            return reply, 1
-        if not args.wait_result:
-            return reply, 0
+    frames = 0
+    if args.record:
+        frames = film(args.record, args.seconds, args.hold, lambda: screenshot(cdp), lambda: latest() is not None)
     else:
-        number = args.id
-        if not (control / f'{number:06d}.request.json').exists():
-            raise ValueError('No submitted request has this ID')
-    reply = wait_reply(control, number, 'result', max(0, deadline-time.monotonic()))
-    return reply, 0 if reply.get('success') is True and reply.get('outcome') == 'completed' else 1
+        deadline = time.monotonic() + args.seconds
+        while latest() is None and time.monotonic() < deadline:
+            time.sleep(0.05)
+    prompt = latest()
+    result = {"outcome": "completed" if prompt else "running"}
+    if prompt:
+        result.update(ok=prompt["ok"], exit_code=prompt["exit_code"], cwd=prompt["cwd"])
+    if args.record:
+        result["frames"] = frames
+        result["scene"] = {"kind": "frames", "src": str(args.record.resolve()), "rate": FPS}
+        write_json(args.record / "take.json", result)
+    print(json.dumps(result, ensure_ascii=False))
+    return 2 if not prompt else 0 if prompt["ok"] else 1
+
+
+def last_prompt_number(directory):
+    reported = prompts(tail(directory / "terminal.log"))
+    return reported[-1]["n"] if reported else 0
+
+
+def run(args):
+    session = load_session(args.session)
+    n0 = last_prompt_number(args.session)
+    cdp = connect(session)
+    type_text(cdp, args.command)
+    press(cdp, "Enter")
+    write_json(args.session / "mark.json", {"n": n0})
+    return observe(args, cdp, n0)
+
+
+def key(args):
+    session = load_session(args.session)
+    n0 = last_prompt_number(args.session)
+    cdp = connect(session)
+    cdp.call("Runtime.evaluate", {"expression": "document.querySelector('.xterm-helper-textarea').focus()"})
+    press(cdp, args.key)
+    write_json(args.session / "mark.json", {"n": n0})
+    return observe(args, cdp, n0)
+
+
+def watch(args):
+    session = load_session(args.session)
+    # Wait for the prompt after the last run/key, even if it already returned.
+    mark = args.session / "mark.json"
+    n0 = read_json(mark)["n"] if mark.exists() else last_prompt_number(args.session)
+    return observe(args, connect(session), n0)
+
+
+def close(args):
+    session = read_json(args.session / "session.json")
+    (args.session / "stop").write_text("")
+    for pid in session["pids"]:
+        kill_process_tree(pid)
+    for _ in range(50):  # the browser releases its profile shortly after dying
+        try:
+            shutil.rmtree(args.session / "profile")
+            break
+        except FileNotFoundError:
+            break
+        except OSError:
+            time.sleep(0.1)
+    print(json.dumps({"closed": True}))
+    return 0
 
 
 def main():
-    if sys.argv[1:] == ['--emit-bash']:
-        emit_bash()
-        return 0
-    parser = argparse.ArgumentParser(description=__doc__)
-    commands = parser.add_subparsers(dest='action', required=True)
-    serve = commands.add_parser('serve')
-    serve.add_argument('--shell', dest='shell_kind', required=True, choices=['powershell51','powershell7','gitbash'])
-    serve.add_argument('--directory', type=Path, required=True)
-    serve.add_argument('--cwd', type=Path, default=Path.cwd())
-    serve.add_argument('--shell-exe')
-    serve.add_argument('--browser')
-    serve.add_argument('--ttyd')
-    request = commands.add_parser('request')
-    request.add_argument('--file', type=Path, required=True)
-    request.add_argument('--wait-result', action='store_true')
-    result = commands.add_parser('result')
-    result.add_argument('--id', type=int, required=True)
-    for command in (request, result):
-        command.add_argument('--directory', type=Path, required=True)
-        command.add_argument('--timeout', type=float, default=30)
+    for stream in (sys.stdout, sys.stderr):
+        if hasattr(stream, "reconfigure"):
+            stream.reconfigure(errors="backslashreplace")
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    verbs = parser.add_subparsers(dest="verb", required=True)
+
+    def filming(sub, seconds):
+        sub.add_argument("--record", type=Path, help="write PNG frames here at 5 fps")
+        sub.add_argument("--seconds", type=float, default=seconds, help="give up waiting after this long")
+        sub.add_argument("--hold", type=float, default=1.5, help="keep filming this long after the prompt returns")
+
+    sub = verbs.add_parser("serve")
+    sub.add_argument("session", type=Path)
+    sub.add_argument("--shell", choices=list(SHELLS), required=True)
+    sub.add_argument("--cwd", type=Path)
+    sub.add_argument("--shell-exe")
+    sub.add_argument("--ttyd")
+    sub.add_argument("--browser")
+    sub = verbs.add_parser("run")
+    sub.add_argument("session", type=Path)
+    sub.add_argument("command")
+    filming(sub, 60)
+    sub = verbs.add_parser("key")
+    sub.add_argument("session", type=Path)
+    sub.add_argument("key")
+    filming(sub, 30)
+    sub = verbs.add_parser("watch")
+    sub.add_argument("session", type=Path)
+    filming(sub, 30)
+    sub = verbs.add_parser("close")
+    sub.add_argument("session", type=Path)
     args = parser.parse_args()
-    try:
-        args.directory = args.directory.resolve()
-        if args.action == 'serve':
-            if sys.platform != 'win32':
-                raise RuntimeError('This example requires native Windows; use the separate Unix tmux recipe')
-            args.cwd = args.cwd.resolve(strict=True)
-            defaults = {'powershell51': str(Path(os.environ['SystemRoot'])/'System32/WindowsPowerShell/v1.0/powershell.exe'),
-                        'powershell7': shutil.which('pwsh.exe'),
-                        'gitbash': shutil.which('bash.exe')}
-            args.shell_exe = args.shell_exe or defaults[args.shell_kind]
-            args.browser = find_browser(args.browser)
-            args.ttyd = args.ttyd or shutil.which('ttyd.exe')
-            for name in ('shell_exe','browser','ttyd'):
-                if not getattr(args,name) or not Path(getattr(args,name)).is_file():
-                    raise FileNotFoundError(f'Provide an existing executable for --{name.replace("_","-")}')
-            Recorder(args).run()
-            return 0
-        if not math.isfinite(args.timeout) or args.timeout < 0:
-            raise ValueError('Client timeout must be finite and nonnegative')
-        reply, code = client(args)
-        print(json.dumps(reply, ensure_ascii=True))
-        return code
-    except (Exception, KeyboardInterrupt) as error:
-        print(json.dumps(dict(success=False, reason=str(error)), ensure_ascii=True), file=sys.stderr)
-        return 1
-
-
-if __name__ == '__main__':
-    raise SystemExit(main())
+    if args.verb == "watch" and not args.record:
+        parser.error("watch needs --record")
+    return {"serve": serve, "run": run, "key": key, "watch": watch, "close": close}[args.verb](args)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

skills/proving-it-works-with-a-movie/recording-a-terminal.md

@@ -4,141 +4,54 @@ CLIs, TUIs, installs, test runs, agents at work — a large share of what is
 worth proving happens in a terminal, and none of it is visible to a browser
 recorder or an OS screen capture you probably can't get permission for.
 
-## Native Windows: one shell, two takes
+## Native Windows: `examples/film-terminal.py` stands in for tmux
 
-`examples/film-terminal.py` serves a native shell through ttyd/ConPTY and an
-owned headless Chrome or Edge page. It needs uv, native Python 3.12+, ttyd,
-and Chrome or Edge. PowerShell itself does not need Bash. Choose the shell
-being recorded with `--shell powershell51|powershell7|gitbash`; the shell
-invoking uv is a separate choice. Use `--shell-exe`, `--ttyd`, and `--browser`
-for explicit executable paths when they are absent from PATH.
+Windows has no tmux, so the example script holds the session instead. `serve`
+starts ttyd on the shell you name and a headless Chrome or Edge page showing
+it, keeps both alive, and appends the raw terminal output to
+`SESSION/terminal.log`. Every other verb is one short call against that
+browser. Run `serve` in a background task your harness keeps alive, the way
+the visual companion server runs; a one-shot shell that kills its children
+on return ends the session.
 
-Run `serve` in a foreground/background task kept alive by your harness,
-like the visual companion. Keep that task running while later shell tool
-calls submit requests. Do not use a one-shot shell that tears down its
-children on return. Do not install a service. PowerShell `Start-Process
--ArgumentList` joins arguments into a string and can lose special-path
-quoting; the foreground invocation below keeps arguments separate.
-
-PowerShell 5.1 or 7, in the long-lived recorder task:
+It needs uv, ttyd, and Chrome or Edge on PATH, or `--ttyd` and `--browser`.
+`--shell powershell51|powershell7|gitbash` picks the filmed shell; the shell
+you type these commands into is a separate choice. From PowerShell:
 
 ```powershell
 $skill = 'C:/path/to/skills/proving-it-works-with-a-movie'
 $work = "$HOME/movie O'Brien λ & [take]"
-& uv run --script "$skill/examples/film-terminal.py" serve --shell powershell51 --directory "$work/session" --cwd "$work"
-```
-
-Git Bash, in the long-lived recorder task:
-
-```bash
-skill=$(cygpath -m '/c/path/to/skills/proving-it-works-with-a-movie')
-work=$(cygpath -m "$HOME/movie O'Brien λ & [take]")
-uv run --script "$skill/examples/film-terminal.py" serve --shell gitbash --directory "$work/session" --cwd "$work"
-```
-
-Create `work` first; use a new session directory each time. Wait for
-`session/control/ready.json` and inspect `session/ready.png`. The ready record
-contains the filmed shell's identity, cwd, geometry, and `next_request_id`.
-Readiness comes through that same filmed terminal; opening a second ttyd
-client would replace the session and is not an observation technique.
-
-For each subsequent PowerShell control call, set `skill` and `work` again
-and define this small request helper. It writes JSON without a BOM and
-preserves each argument:
-
-```powershell
-function Send-MovieRequest([hashtable]$data, [switch]$WaitResult) {
-    $path = "$work/request-$($data.id).json"
-    $json = $data | ConvertTo-Json -Compress
-    [IO.File]::WriteAllText($path, $json, [Text.UTF8Encoding]::new($false))
-    $arguments = @('run','--script',"$skill/examples/film-terminal.py",'request',
-        '--directory',"$work/session",'--file',$path)
-    if ($WaitResult) { $arguments += '--wait-result' }
-    & uv @arguments
-    if ($LASTEXITCODE -ne 0) { throw 'Recorder request failed' }
-}
-# First control call: a real command keeps stdin until the later Enter key.
-Send-MovieRequest @{id=1;operation='begin-take';name='take-one'} -WaitResult
-Send-MovieRequest @{id=2;operation='run';command='python -u -c "print(123); input(); print(456)"';native_producer='python';timeout_seconds=120}
-Start-Sleep -Seconds 2
-Send-MovieRequest @{id=3;operation='end-take'} -WaitResult
-```
-
-In a **later control call**, with the same paths/helper and server still alive:
-
-```powershell
-Send-MovieRequest @{id=4;operation='begin-take';name='take-two'} -WaitResult
-Send-MovieRequest @{id=5;operation='key';key='Enter'} -WaitResult
-& uv run --script "$skill/examples/film-terminal.py" result --directory "$work/session" --id 2 --timeout 30
-if ($LASTEXITCODE -ne 0) { throw 'Recorded command did not succeed' }
-Start-Sleep -Seconds 2  # Keep the observed completion readable in the movie.
-Send-MovieRequest @{id=6;operation='end-take'} -WaitResult
-Send-MovieRequest @{id=7;operation='close'} -WaitResult
-```
-
-Equivalent Git Bash control calls use UTF-8 files and native-form paths.
-Set `skill` and `work` in each call. The first call:
-
-```bash
-set -euo pipefail
-recorder="$skill/examples/film-terminal.py"
-printf '%s\n' '{"id":1,"operation":"begin-take","name":"take-one"}' > "$work/1.json"
-printf '%s\n' '{"id":2,"operation":"run","command":"python -u -c \"print(123); input(); print(456)\"","native_producer":"python","timeout_seconds":120}' > "$work/2.json"
-printf '%s\n' '{"id":3,"operation":"end-take"}' > "$work/3.json"
-uv run --script "$recorder" request --directory "$work/session" --file "$work/1.json" --wait-result
-uv run --script "$recorder" request --directory "$work/session" --file "$work/2.json"
-sleep 2
-uv run --script "$recorder" request --directory "$work/session" --file "$work/3.json" --wait-result
-```
-
-The later Git Bash call:
-
-```bash
-set -euo pipefail
-recorder="$skill/examples/film-terminal.py"
-printf '%s\n' '{"id":4,"operation":"begin-take","name":"take-two"}' > "$work/4.json"
-printf '%s\n' '{"id":5,"operation":"key","key":"Enter"}' > "$work/5.json"
-printf '%s\n' '{"id":6,"operation":"end-take"}' > "$work/6.json"
-printf '%s\n' '{"id":7,"operation":"close"}' > "$work/7.json"
-uv run --script "$recorder" request --directory "$work/session" --file "$work/4.json" --wait-result
-uv run --script "$recorder" request --directory "$work/session" --file "$work/5.json" --wait-result
-uv run --script "$recorder" result --directory "$work/session" --id 2 --timeout 30
-sleep 2
-uv run --script "$recorder" request --directory "$work/session" --file "$work/6.json" --wait-result
-uv run --script "$recorder" request --directory "$work/session" --file "$work/7.json" --wait-result
+$film = "$skill/examples/film-terminal.py"
+# in a background task, kept alive until close:
+& uv run --script $film serve "$work/session" --shell powershell7 --cwd $work
+# then one call each; wait for "$work/session/ready.json" first:
+& uv run --script $film run "$work/session" 'pytest -q' --record "$work/take-one"
+& uv run --script $film run "$work/session" 'python app.py' --record "$work/take-two" --seconds 8  # a TUI: exits 2, still running
+& uv run --script $film key "$work/session" q --record "$work/take-three"
+& uv run --script $film close "$work/session"
 ```
 
-Use one controller and consecutive IDs starting at one. Acknowledgment only
-means accepted. `--wait-result` and the wait-only `result` command return
-nonzero on failed, unknown, interrupted, or missing results. After a client
-wait timeout, retrieve the original ID with `result`; do not submit it again.
-The command's `timeout_seconds` is separate: its expiry ends the session and
-marks the outcome unknown. `inspect` reports the pending command, active
-take, and next ID; it consumes an ID like every other request.
-
-`end-take` stops capture, preserving shell variables, cwd, and a pending
-command. A second `run` is rejected while one is pending. Use intentional
-`key` requests for input: printable characters (such as the fixture's `q`),
-Enter, Escape, arrows, Tab, and Ctrl-C. `close` finalizes a healthy take;
-`cancel` marks it incomplete. Both release only owned processes and mark a
-pending command interrupted. Wait for the shutdown result before declaring
-cleanup successful.
-
-The viewport is fixed at 1600×900, sampled at a target 5 fps. Read the actual
-rows/columns from readiness; resizing fails the session. Hold important
-states at least 1.3 seconds. A screenshot gap above two seconds fails a take.
-Exports use completed samples on the 0.2-second grid and record duplicates
-in `take.json`; sampling does not prove every faster event was observed.
-Each completed `end-take` supplies `kind: frames`, `src`, and `rate: 5` for
-assembling.md. Keep the final movie at a readable resolution and inspect the
-exported frames and finished movie, including command completion.
-
-For native commands followed by logging, set `native_producer` to the
-explicit first-stage executable. This prevents successful `tee`/`Tee-Object`
-from hiding producer failure. Omit it for opaque/mixed scripts; their result
-does not certify every internal command. Preserve raw PowerShell `$?`
-separately from request-attributed errors and native exit status. See
-rendering-from-a-log.md for direct shell logging recipes.
+From Git Bash the commands are the same with `skill=$(cygpath -m ...)` and
+`work=$(cygpath -m ...)`, as in assembling.md.
+
+`run` types the command and films at 5 fps into `--record` until the prompt
+comes back, holds 1.5 s so the result stays readable, and prints the status
+as JSON: `ok` is the shell's own success flag and `exit_code` the last native
+program's exit code, which PowerShell keeps from an earlier program when the
+command was a cmdlet. It exits 1 when the command failed and 2 when it is
+still running after `--seconds`. `key` presses one key and `watch` films
+without typing; both wait for the prompt the same way. Every `--record`
+directory is a `kind: frames` scene at `rate: 5`; a slow screenshot repeats
+the previous frame so the timing stays honest.
+
+Long work spans takes exactly as on Unix: film the command being issued with
+a short `--seconds`, do other things, then `watch` the result as a new take.
+The shell, its variables and its cwd persist across calls until `close`,
+which kills ttyd, the browser and everything they started.
+
+The viewport is fixed at 1600×900 with a 17 px font. Look at
+`SESSION/ready.png` before filming; `serve` refuses a blank canvas, the same
+preflight as below.
 
 ## Unix: tmux and ttyd
 

skills/proving-it-works-with-a-movie/scripts/browser_tools.py

@@ -1,4 +1,4 @@
-"""Owned headless browser discovery and bounded card screenshots."""
+"""Headless browser discovery, bounded card screenshots, and process-tree cleanup."""
 
 from __future__ import annotations
 
@@ -11,8 +11,6 @@ import tempfile
 import time
 from pathlib import Path
 
-from windows_jobs import WindowsJob
-
 
 UNIX_BROWSERS = [
     "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
@@ -59,6 +57,29 @@ def find_browser(explicit: str | None) -> str | None:
     return None
 
 
+def _descendants(pid: int) -> list[int]:
+    pids, index = [pid], 0
+    while index < len(pids):
+        listed = subprocess.run(["pgrep", "-P", str(pids[index])], capture_output=True, text=True).stdout
+        pids.extend(int(child) for child in listed.split())
+        index += 1
+    return pids
+
+
+def kill_process_tree(pid: int) -> None:
+    """Kill a process this tool started and everything it spawned. A browser
+    or ttyd leaves helpers behind otherwise, and on Unix a pty child starts
+    its own session, so a process group is not enough."""
+    if sys.platform == "win32":
+        subprocess.run(["taskkill", "/T", "/F", "/PID", str(pid)], capture_output=True)
+        return
+    for victim in reversed(_descendants(pid)):
+        try:
+            os.kill(victim, signal.SIGKILL)
+        except ProcessLookupError:
+            pass
+
+
 def render_card(html: Path, png: Path, *, browser: str, width: int,
                 height: int, timeout: float = 20) -> None:
     """Render one local HTML page and release every process it launched."""
@@ -68,53 +89,39 @@ def render_card(html: Path, png: Path, *, browser: str, width: int,
         raise FileNotFoundError(f"card HTML does not exist: {html}")
     png.parent.mkdir(parents=True, exist_ok=True)
     png.unlink(missing_ok=True)
-    with tempfile.TemporaryDirectory(prefix="movie-browser-") as profile:
+    with tempfile.TemporaryDirectory(prefix="movie-browser-", ignore_cleanup_errors=True) as profile:
         profile_path = Path(profile)
         log = profile_path / "browser.log"
-        job = None
-        process = None
+        argv = [
+            str(Path(browser).resolve()) if Path(browser).is_file() else browser,
+            "--headless=new", "--disable-gpu", "--hide-scrollbars",
+            "--no-first-run", "--no-default-browser-check",
+            f"--user-data-dir={profile_path}", f"--screenshot={png}",
+            f"--window-size={width},{height}", "--force-device-scale-factor=1",
+            html.as_uri(),
+        ]
+        with log.open("wb") as output:
+            process = subprocess.Popen(argv, cwd=profile_path, stdin=subprocess.DEVNULL,
+                                       stdout=output, stderr=subprocess.STDOUT,
+                                       start_new_session=sys.platform != "win32")
         try:
-            argv = [
-                str(Path(browser).resolve()) if Path(browser).is_file() else browser,
-                "--headless=new", "--disable-gpu", "--hide-scrollbars",
-                "--no-first-run", "--no-default-browser-check",
-                f"--user-data-dir={profile_path}", f"--screenshot={png}",
-                f"--window-size={width},{height}", "--force-device-scale-factor=1",
-                html.as_uri(),
-            ]
-            with log.open("wb") as output:
-                if sys.platform == "win32":
-                    job = WindowsJob()
-                    pid = job.spawn(argv, profile_path, log)
-                else:
-                    process = subprocess.Popen(argv, cwd=profile_path,
-                        stdin=subprocess.DEVNULL, stdout=output, stderr=subprocess.STDOUT,
-                        start_new_session=True)
-                deadline = time.monotonic() + timeout
-                while time.monotonic() < deadline:
-                    # A fresh profile may keep background services alive after
-                    # taking the screenshot. A complete PNG is the render result.
-                    if png.is_file():
-                        data = png.read_bytes()
-                        if data.startswith(b"\x89PNG\r\n\x1a\n") and data.endswith(b"IEND\xaeB`\x82"):
-                            return
-                    try:
-                        remaining = min(0.05, max(0, deadline - time.monotonic()))
-                        code = job.wait(pid, remaining) if job else process.wait(timeout=remaining)
-                    except (TimeoutError, subprocess.TimeoutExpired):
-                        continue
-                    if code != 0 or not png.is_file() or png.stat().st_size == 0:
-                        detail = log.read_text(encoding="utf-8", errors="replace")[-1000:]
-                        raise RuntimeError(f"Browser exited with status {code} without a complete PNG: {detail}")
-                    # Read the file on the next iteration after a successful exit.
-                    time.sleep(0.01)
-                raise TimeoutError(f"Browser exceeded {timeout:g}s")
+            deadline = time.monotonic() + timeout
+            while time.monotonic() < deadline:
+                # A fresh profile may keep background services alive after
+                # taking the screenshot. A complete PNG is the render result.
+                if png.is_file():
+                    data = png.read_bytes()
+                    if data.startswith(b"\x89PNG\r\n\x1a\n") and data.endswith(b"IEND\xaeB`\x82"):
+                        return
+                elif process.poll() is not None:
+                    detail = log.read_text(encoding="utf-8", errors="replace")[-1000:]
+                    raise RuntimeError(f"Browser exited with status {process.returncode} "
+                                       f"without a complete PNG: {detail}")
+                time.sleep(0.05)
+            raise TimeoutError(f"Browser exceeded {timeout:g}s")
         finally:
-            if job is not None:
-                job.close()
-            if process is not None:
-                try:
-                    os.killpg(process.pid, signal.SIGKILL)
-                except ProcessLookupError:
-                    pass
+            kill_process_tree(process.pid)
+            try:
                 process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                pass

skills/proving-it-works-with-a-movie/scripts/windows_jobs.py

@@ -1,236 +0,0 @@
-"""Owned Windows process trees for movie capture, extracted from the native probe."""
-
-import os
-import subprocess
-import sys
-import time
-from pathlib import Path
-
-
-class WindowsJob:
-    """Suspended roots enter an unnamed, non-inheritable job before resuming."""
-
-    def __init__(self):
-        if sys.platform != "win32":
-            raise RuntimeError("WindowsJob requires native Windows")
-        import ctypes
-        from ctypes import wintypes as W
-        self.ctypes = ctypes
-        U64, SIZE = ctypes.c_ulonglong, ctypes.c_size_t
-
-        class STARTUPINFOW(ctypes.Structure):
-            _fields_ = [("cb", W.DWORD), ("lpReserved", W.LPWSTR),
-                        ("lpDesktop", W.LPWSTR), ("lpTitle", W.LPWSTR),
-                        ("dwX", W.DWORD), ("dwY", W.DWORD), ("dwXSize", W.DWORD),
-                        ("dwYSize", W.DWORD), ("dwXCountChars", W.DWORD),
-                        ("dwYCountChars", W.DWORD), ("dwFillAttribute", W.DWORD),
-                        ("dwFlags", W.DWORD), ("wShowWindow", W.WORD),
-                        ("cbReserved2", W.WORD), ("lpReserved2", ctypes.POINTER(W.BYTE)),
-                        ("hStdInput", W.HANDLE), ("hStdOutput", W.HANDLE), ("hStdError", W.HANDLE)]
-
-        class PROCESS_INFORMATION(ctypes.Structure):
-            _fields_ = [("hProcess", W.HANDLE), ("hThread", W.HANDLE),
-                        ("dwProcessId", W.DWORD), ("dwThreadId", W.DWORD)]
-
-        class BASIC_LIMIT(ctypes.Structure):
-            _fields_ = [("PerProcessUserTimeLimit", ctypes.c_longlong),
-                        ("PerJobUserTimeLimit", ctypes.c_longlong), ("LimitFlags", W.DWORD),
-                        ("MinimumWorkingSetSize", SIZE), ("MaximumWorkingSetSize", SIZE),
-                        ("ActiveProcessLimit", W.DWORD), ("Affinity", SIZE),
-                        ("PriorityClass", W.DWORD), ("SchedulingClass", W.DWORD)]
-
-        class IO_COUNTERS(ctypes.Structure):
-            _fields_ = [(name, U64) for name in ("ReadOperationCount", "WriteOperationCount",
-                        "OtherOperationCount", "ReadTransferCount", "WriteTransferCount", "OtherTransferCount")]
-
-        class EXTENDED_LIMIT(ctypes.Structure):
-            _fields_ = [("BasicLimitInformation", BASIC_LIMIT), ("IoInfo", IO_COUNTERS),
-                        ("ProcessMemoryLimit", SIZE), ("JobMemoryLimit", SIZE),
-                        ("PeakProcessMemoryUsed", SIZE), ("PeakJobMemoryUsed", SIZE)]
-
-        self.sizes = {"STARTUPINFOW": ctypes.sizeof(STARTUPINFOW),
-                      "PROCESS_INFORMATION": ctypes.sizeof(PROCESS_INFORMATION),
-                      "BASIC_LIMIT": ctypes.sizeof(BASIC_LIMIT),
-                      "IO_COUNTERS": ctypes.sizeof(IO_COUNTERS),
-                      "EXTENDED_LIMIT": ctypes.sizeof(EXTENDED_LIMIT)}
-        self.SI, self.PI = STARTUPINFOW, PROCESS_INFORMATION
-        self.k = ctypes.WinDLL("kernel32", use_last_error=True)
-        signatures = {
-            "CreateJobObjectW": ([ctypes.c_void_p, W.LPCWSTR], W.HANDLE),
-            "SetInformationJobObject": ([W.HANDLE, ctypes.c_int, ctypes.c_void_p, W.DWORD], W.BOOL),
-            "QueryInformationJobObject": ([W.HANDLE, ctypes.c_int, ctypes.c_void_p, W.DWORD, ctypes.POINTER(W.DWORD)], W.BOOL),
-            "CreateProcessW": ([W.LPCWSTR, W.LPWSTR, ctypes.c_void_p, ctypes.c_void_p, W.BOOL,
-                                W.DWORD, ctypes.c_void_p, W.LPCWSTR, ctypes.POINTER(STARTUPINFOW),
-                                ctypes.POINTER(PROCESS_INFORMATION)], W.BOOL),
-            "AssignProcessToJobObject": ([W.HANDLE, W.HANDLE], W.BOOL),
-            "ResumeThread": ([W.HANDLE], W.DWORD),
-            "TerminateProcess": ([W.HANDLE, W.UINT], W.BOOL),
-            "TerminateJobObject": ([W.HANDLE, W.UINT], W.BOOL),
-            "CloseHandle": ([W.HANDLE], W.BOOL),
-            "WaitForSingleObject": ([W.HANDLE, W.DWORD], W.DWORD),
-            "GetExitCodeProcess": ([W.HANDLE, ctypes.POINTER(W.DWORD)], W.BOOL),
-            "OpenProcess": ([W.DWORD, W.BOOL, W.DWORD], W.HANDLE),
-            "GetProcessTimes": ([W.HANDLE] + [ctypes.POINTER(W.FILETIME)] * 4, W.BOOL),
-            "IsProcessInJob": ([W.HANDLE, W.HANDLE, ctypes.POINTER(W.BOOL)], W.BOOL),
-        }
-        for name, (arguments, returns) in signatures.items():
-            function = getattr(self.k, name)
-            function.argtypes, function.restype = arguments, returns
-        self.handle = self.k.CreateJobObjectW(None, None)
-        if not self.handle:
-            raise ctypes.WinError(ctypes.get_last_error())
-        limits = EXTENDED_LIMIT()
-        limits.BasicLimitInformation.LimitFlags = 0x2000  # KILL_ON_JOB_CLOSE; no breakaway
-        if not self.k.SetInformationJobObject(self.handle, 9, ctypes.byref(limits), ctypes.sizeof(limits)):
-            error = ctypes.get_last_error()
-            self.k.CloseHandle(self.handle)
-            raise ctypes.WinError(error)
-        self.roots = {}
-
-    def spawn(self, argv: list[str], directory: Path, log: Path,
-              env: dict[str, str] | None = None) -> int:
-        import msvcrt
-        ctypes = self.ctypes
-        if not self.handle:
-            raise RuntimeError("Job is closed")
-
-        if not argv or not Path(argv[0]).is_absolute():
-            raise ValueError("An absolute executable path is required")
-        environment = dict(os.environ) if env is None else env
-        block = ctypes.create_unicode_buffer("\0".join(f"{k}={v}" for k, v in sorted(environment.items())) + "\0\0")
-        si, pi = self.SI(), self.PI()
-        si.cb, si.dwFlags = ctypes.sizeof(si), 0x100  # STARTF_USESTDHANDLES
-        with open(os.devnull, "rb") as stdin, log.open("ab", buffering=0) as output:
-            handles = [msvcrt.get_osfhandle(f.fileno()) for f in (stdin, output)]
-            for handle in handles:
-                os.set_handle_inheritable(handle, True)
-            si.hStdInput, si.hStdOutput, si.hStdError = handles[0], handles[1], handles[1]
-            try:
-                created = self.k.CreateProcessW(argv[0], ctypes.create_unicode_buffer(subprocess.list2cmdline(argv)),
-                    None, None, True, 0x404, block, str(directory), ctypes.byref(si), ctypes.byref(pi))
-            finally:
-                for handle in handles:
-                    os.set_handle_inheritable(handle, False)
-        if not created:
-            raise ctypes.WinError(ctypes.get_last_error())
-        try:
-            if not self.k.AssignProcessToJobObject(self.handle, pi.hProcess):
-                raise ctypes.WinError(ctypes.get_last_error())
-            if self.k.ResumeThread(pi.hThread) == 0xFFFFFFFF:
-                raise ctypes.WinError(ctypes.get_last_error())
-        except BaseException:
-            self.k.TerminateProcess(pi.hProcess, 1)
-            self.k.WaitForSingleObject(pi.hProcess, 5000)
-            self.k.CloseHandle(pi.hProcess)
-            raise
-        finally:
-            self.k.CloseHandle(pi.hThread)
-        self.roots[pi.dwProcessId] = pi.hProcess
-        return pi.dwProcessId
-
-    def wait(self, pid: int, timeout: float) -> int:
-        handle = self.roots[pid]
-        result = self.k.WaitForSingleObject(handle, max(0, int(timeout * 1000)))
-        if result == 258:
-            raise TimeoutError(f"Owned process {pid} exceeded {timeout:g}s")
-        if result != 0:
-            raise self.ctypes.WinError(self.ctypes.get_last_error())
-        from ctypes import wintypes as W
-        code = W.DWORD()
-        if not self.k.GetExitCodeProcess(handle, self.ctypes.byref(code)):
-            raise self.ctypes.WinError(self.ctypes.get_last_error())
-        return int(code.value)
-
-    def pids(self):
-        ctypes = self.ctypes
-        from ctypes import wintypes as W
-
-        class PROCESS_LIST(ctypes.Structure):
-            _fields_ = [("assigned", W.DWORD), ("count", W.DWORD), ("pids", ctypes.c_size_t * 1024)]
-
-        info = PROCESS_LIST()
-        if not self.k.QueryInformationJobObject(self.handle, 3, ctypes.byref(info), ctypes.sizeof(info), None):
-            raise ctypes.WinError(ctypes.get_last_error())
-        return list(info.pids[:info.count])
-
-    def process_time(self, handle):
-        ctypes = self.ctypes
-        from ctypes import wintypes as W
-
-        times = [W.FILETIME() for _ in range(4)]
-        if not self.k.GetProcessTimes(handle, *(ctypes.byref(value) for value in times)):
-            raise ctypes.WinError(ctypes.get_last_error())
-        return (times[0].dwHighDateTime << 32) | times[0].dwLowDateTime
-
-    def open_process(self, pid, creation=None, terminate=False):
-        ctypes = self.ctypes
-        handle = self.k.OpenProcess(0x100000 | 0x1000 | int(terminate), False, pid)
-        if not handle:
-            raise ctypes.WinError(ctypes.get_last_error())
-        try:
-            if creation is not None and self.process_time(handle) != creation:
-                raise RuntimeError("Process creation time changed; refusing stale PID")
-        except BaseException:
-            self.k.CloseHandle(handle)
-            raise
-        return handle
-
-    def snapshot(self):
-        ctypes = self.ctypes
-        from ctypes import wintypes as W
-
-        processes = []
-        try:
-            for pid in self.pids():
-                try:
-                    handle = self.open_process(pid)
-                except OSError as error:
-                    if error.winerror == 87:  # Exited between enumeration and open.
-                        continue
-                    raise
-                try:
-                    owned = W.BOOL()
-                    if not self.k.IsProcessInJob(handle, self.handle, ctypes.byref(owned)) or not owned.value:
-                        raise RuntimeError("Process is no longer a member of the owned job")
-                    creation = self.process_time(handle)
-                except BaseException:
-                    self.k.CloseHandle(handle)
-                    raise
-                processes.append({"pid": pid, "handle": handle, "creation": creation})
-            return processes
-        except BaseException:
-            for process in processes:
-                self.k.CloseHandle(process["handle"])
-            raise
-
-    def close(self) -> None:
-        ctypes = self.ctypes
-        if not self.handle:
-            return
-        processes = []
-        try:
-            # The job's active PID list can empty before terminated processes
-            # finish releasing files. Retain identities and wait for exit too.
-            processes = self.snapshot()
-            if not self.k.TerminateJobObject(self.handle, 1):
-                raise ctypes.WinError(ctypes.get_last_error())
-            deadline = time.monotonic() + 5
-            for process in processes:
-                remaining = max(0, int((deadline - time.monotonic()) * 1000))
-                status = self.k.WaitForSingleObject(process["handle"], remaining)
-                if status == 258:
-                    raise TimeoutError("Owned process did not finish termination")
-                if status != 0:
-                    raise ctypes.WinError(ctypes.get_last_error())
-            while self.pids() and time.monotonic() < deadline:
-                time.sleep(0.05)
-            if self.pids():
-                raise TimeoutError("Owned processes remain after job termination")
-        finally:
-            self.k.CloseHandle(self.handle)
-            self.handle = None
-            for process in processes:
-                self.k.CloseHandle(process["handle"])
-            for handle in self.roots.values():
-                self.k.CloseHandle(handle)
-            self.roots.clear()

tests/proving-it-works-with-a-movie/README.md

@@ -16,8 +16,9 @@ them; any skip then makes the run fail.
 The assembly sine wave is only a synthetic timing fixture. The narration drift
 inputs exercise text comparison only. Neither is speech/ASR acceptance.
 
-The `processes` and `terminal` suites need native Windows with ttyd and Chrome
-or Edge available; elsewhere they skip. Set `MOVIE_TEST_SHELL` to
-`powershell51`, `powershell7`, or `gitbash` to choose the recorded shell, and
-`MOVIE_TEST_SHELL_EXE` and `MOVIE_TEST_TTYD` to name those executables when
-they are not on PATH.
+The `terminal` suite starts a real ttyd session and skips where ttyd or a
+Chrome-family browser is missing. It films `bash` by default on Unix and
+`powershell51` on Windows; set `MOVIE_TEST_SHELL` to `powershell51`,
+`powershell7`, `gitbash`, or `bash`, and `MOVIE_TEST_SHELL_EXE`,
+`MOVIE_TEST_TTYD`, or `MOVIE_TEST_BROWSER` when those executables are not on
+PATH.

tests/proving-it-works-with-a-movie/fixtures/terminal_app.py

@@ -1,22 +1,41 @@
-"""Native wrapping/TUI fixture and descendant tree for recorder ownership tests."""
-import os, subprocess, sys
+"""TUI fixture: three timed colour states, then wait for `q`.
+`tree DIR` instead leaves a three-deep process tree running for cleanup tests."""
+import json, os, subprocess, sys, time
 from pathlib import Path
-if len(sys.argv)>1 and sys.argv[1]=='tree':
-    import json,time
-    directory=Path(sys.argv[2]);directory.mkdir(exist_ok=True)
-    level=int(sys.argv[3]) if len(sys.argv)>3 else 0
-    if level<2:subprocess.Popen([sys.executable,__file__,'tree',str(directory),str(level+1)])
-    (directory/f'{level}.json').write_text(json.dumps(dict(pid=os.getpid(),level=level)))
-    while True:time.sleep(1)
-import json, msvcrt, time
-from pathlib import Path
-print('WRAPPING OUTPUT '+('0123456789 wrap proof '*60),flush=True)
+
+
+def getch():
+    if os.name == "nt":
+        import msvcrt
+        return msvcrt.getwch()
+    import termios, tty
+    fd = sys.stdin.fileno()
+    old = termios.tcgetattr(fd)
+    try:
+        tty.setcbreak(fd)
+        return sys.stdin.read(1)
+    finally:
+        termios.tcsetattr(fd, termios.TCSADRAIN, old)
+
+
+if len(sys.argv) > 1 and sys.argv[1] == "tree":
+    directory = Path(sys.argv[2])
+    directory.mkdir(exist_ok=True)
+    level = int(sys.argv[3]) if len(sys.argv) > 3 else 0
+    if level < 2:
+        subprocess.Popen([sys.executable, __file__, "tree", str(directory), str(level + 1)])
+    (directory / f"{level}.json").write_text(json.dumps(dict(pid=os.getpid(), level=level)))
+    while True:
+        time.sleep(1)
+
+print("WRAPPING OUTPUT " + ("0123456789 wrap proof " * 60), flush=True)
 time.sleep(1.4)
-states=[]
-for label,color in [('STATE ONE RED',41),('STATE TWO GREEN',42),('STATE THREE BLUE',44)]:
-    print('\x1b[2J\x1b[H'+f'\x1b[{color}m'+(label+' '*60+'\n')*12+'\x1b[0m',end='',flush=True)
-    states.append(dict(label=label,time=time.monotonic()))
-    Path('states.json').write_text(json.dumps(states))
+states = []
+for label, color in [("STATE ONE RED", 41), ("STATE TWO GREEN", 42), ("STATE THREE BLUE", 44)]:
+    print("\x1b[2J\x1b[H" + f"\x1b[{color}m" + (label + " " * 60 + "\n") * 12 + "\x1b[0m", end="", flush=True)
+    states.append(dict(label=label, time=time.monotonic()))
+    Path("states.json").write_text(json.dumps(states))
     time.sleep(1.5)
-while msvcrt.getwch()!='q': pass
-print('\nAUTOMATIC GATE COMPLETE',flush=True)
+while getch() != "q":
+    pass
+print("\nAUTOMATIC GATE COMPLETE", flush=True)

tests/proving-it-works-with-a-movie/run-tests.py

@@ -17,7 +17,6 @@ IMPLEMENTED_SUITES = {
     "checker": "test_checker.py",
     "narration": "test_narration.py",
     "paths": "test_paths.py",
-    "processes": "test_processes.py",
     "subtitles": "test_subtitles.py",
     "terminal": "test_terminal.py",
 }

tests/proving-it-works-with-a-movie/test_processes.py

@@ -1,99 +0,0 @@
-import os
-import tempfile
-import unittest
-from pathlib import Path
-
-import fixtures
-
-
-class IdentityFailureTests(unittest.TestCase):
-    def job(self):
-        import ctypes
-        from unittest.mock import Mock
-        module = fixtures.load_script("windows_jobs")
-        job = module.WindowsJob.__new__(module.WindowsJob)
-        job.ctypes = ctypes
-        job.k = Mock()
-        job.handle = 10
-        return job
-
-    def test_open_identity_failure_closes_current_handle(self):
-        from unittest.mock import Mock
-        job = self.job()
-        job.k.OpenProcess.return_value = 21
-        job.process_time = Mock(side_effect=RuntimeError("identity unavailable"))
-        with self.assertRaisesRegex(RuntimeError, "identity unavailable"):
-            job.open_process(3, creation=123)
-        job.k.CloseHandle.assert_called_once_with(21)
-
-    def test_snapshot_identity_failure_closes_current_and_previous_handles(self):
-        from unittest.mock import Mock, call
-        job = self.job()
-        job.pids = Mock(return_value=[1, 2])
-        job.open_process = Mock(side_effect=[21, 22])
-        def owned(handle, parent, output):
-            output._obj.value = True
-            return True
-        job.k.IsProcessInJob.side_effect = owned
-        job.process_time = Mock(side_effect=[123, RuntimeError("identity unavailable")])
-        with self.assertRaisesRegex(RuntimeError, "identity unavailable"):
-            job.snapshot()
-        self.assertCountEqual(job.k.CloseHandle.call_args_list, [call(21), call(22)])
-
-
-@unittest.skipUnless(os.name == "nt", "Windows Job ownership is native Windows only")
-class WindowsJobRegression(unittest.TestCase):
-    def test_job_wait_and_close_own_child_process(self):
-        module = fixtures.load_script("windows_jobs")
-        with tempfile.TemporaryDirectory() as directory:
-            root = Path(directory)
-            log = root / "child.log"
-            job = module.WindowsJob()
-            try:
-                pid = job.spawn([str(Path(os.environ["ComSpec"])), "/c", "exit 7"], root, log)
-                self.assertEqual(job.wait(pid, 10), 7)
-            finally:
-                job.close()
-
-@unittest.skipUnless(os.name == "nt", "Windows Job ownership is native Windows only")
-class WindowsJobCleanupRegression(unittest.TestCase):
-    def test_timeout_closes_descendants_without_touching_sentinel(self):
-        import subprocess
-        import sys
-        import time
-        module = fixtures.load_script("windows_jobs")
-        sentinel = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
-        try:
-            with tempfile.TemporaryDirectory() as directory:
-                root = Path(directory)
-                child = "import time; time.sleep(60)"
-                code = f"import subprocess,sys,time; subprocess.Popen([sys.executable, '-c', {child!r}]); time.sleep(60)"
-                job = module.WindowsJob()
-                snapshot = []
-                try:
-                    pid = job.spawn([sys.executable, "-c", code], root, root / "log")
-                    with self.assertRaises(TimeoutError):
-                        job.wait(pid, 0.05)
-                    deadline = time.monotonic() + 5
-                    while len(job.pids()) < 2 and time.monotonic() < deadline:
-                        time.sleep(0.05)
-                    snapshot = job.snapshot()
-                    self.assertGreaterEqual(len(snapshot), 2)
-                    start = time.monotonic()
-                    job.close()
-                    job.close()
-                    self.assertLess(time.monotonic() - start, 10)
-                    for process in snapshot:
-                        self.assertEqual(job.k.WaitForSingleObject(process["handle"], 0), 0)
-                    self.assertIsNone(sentinel.poll())
-                finally:
-                    job.close()
-                    for process in snapshot:
-                        job.k.CloseHandle(process["handle"])
-        finally:
-            sentinel.terminate()
-            sentinel.wait(timeout=10)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/proving-it-works-with-a-movie/test_terminal.py

@@ -1,490 +1,248 @@
-"""Windows terminal recorder timing, control, and native acceptance."""
+"""The terminal recorder: prompt parsing and the frame grid anywhere; a real
+ttyd session wherever ttyd and a Chrome-family browser exist."""
 import importlib.util
-from pathlib import Path
-import unittest
-
-SCRIPT=Path(__file__).resolve().parents[2]/'skills/proving-it-works-with-a-movie/examples/film-terminal.py'
-
-class TerminalPolicyTests(unittest.TestCase):
-    def recorder(self):
-        self.assertTrue(SCRIPT.is_file(), 'Windows recorder entry point is missing')
-        spec=importlib.util.spec_from_file_location('film_terminal',SCRIPT)
-        module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
-        return module
-
-    def test_output_grid_never_uses_future_capture(self):
-        m=self.recorder()
-        self.assertEqual(m.frame_sources([10.,10.35,10.81],10.,11.),[0,0,1,1,1])
-        self.assertEqual(m.frame_sources([10.],10.,10.),[0])
-        self.assertEqual(m.frame_sources([10.,10.2],10.,10.4),[0,1,1])
-
-    def test_capture_gaps_and_invalid_boundaries_fail(self):
-        m=self.recorder()
-        for times,start,end in [([],0,1),([1],0,2),([1],1,.9),([1,3,2],1,4),([1,4],1,3)]:
-            with self.subTest(times=times),self.assertRaises(ValueError):m.frame_sources(times,start,end)
-        for times,start,end in [([1,3.01],1,4),([1],1,3.01)]:
-            with self.subTest(times=times),self.assertRaises(TimeoutError):m.frame_sources(times,start,end)
-
-    def test_logger_cannot_hide_identified_producer_failure(self):
-        m=self.recorder()
-        good=dict(outcome='completed',shell_success=True,shell_error=None,native_producer='python.exe',producer_exit_code=0)
-        self.assertTrue(m.command_succeeded(good))
-        for change in [dict(producer_exit_code=7),dict(producer_exit_code=None),dict(outcome='unknown'),dict(outcome='interrupted'),dict(shell_success=None),dict(shell_error='failed')]:
-            with self.subTest(change=change):self.assertFalse(m.command_succeeded(good|change))
-        self.assertTrue(m.command_succeeded(good|dict(native_producer=None,producer_exit_code=None)))
-
-class ReadinessDirectoryTests(unittest.TestCase):
-    def test_readiness_requires_requested_directory(self):
-        import tempfile
-        from types import SimpleNamespace
-        from unittest.mock import Mock
-        module = TerminalPolicyTests().recorder()
-        with tempfile.TemporaryDirectory() as tmp:
-            cwd = (Path(tmp) / "movie O'Brien λ & [take]").resolve()
-            cwd.mkdir()
-            for shell in ['powershell51', 'powershell7', 'gitbash']:
-                for actual in [cwd, Path(tmp)]:
-                    with self.subTest(shell=shell, actual=actual):
-                        terminal = module.Terminal.__new__(module.Terminal)
-                        terminal.args = SimpleNamespace(shell_kind=shell, cwd=cwd)
-                        terminal.session = 'session'
-                        terminal.directory = Path(tmp)
-                        terminal.closed = False
-                        terminal.parser = SimpleNamespace(records=[{'session':'session', 'cwd':str(actual)}])
-                        terminal.cdp = SimpleNamespace(pump=Mock())
-                        terminal.type = Mock()
-                        terminal.screenshot = Mock()
-                        if actual == cwd:
-                            self.assertEqual(terminal.readiness()['cwd'], str(cwd))
-                            terminal.screenshot.assert_called_once_with('ready.png')
-                        else:
-                            with self.assertRaisesRegex(RuntimeError, 'requested cwd'):
-                                terminal.readiness()
-                            terminal.screenshot.assert_not_called()
-
-
 import json
 import os
+import shlex
+import shutil
 import subprocess
 import sys
 import tempfile
 import time
+import unittest
+from pathlib import Path
 
-@unittest.skipUnless(sys.platform == 'win32', 'native Windows required')
-class NativeCwdFailureTests(unittest.TestCase):
-    def test_missing_cwd_fails_before_launch(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            directory = Path(tmp) / 'session'
-            result = subprocess.run([sys.executable, str(SCRIPT), 'serve', '--shell',
-                'powershell51', '--directory', str(directory), '--cwd', str(Path(tmp) / 'missing')],
-                capture_output=True, timeout=10)
-            self.assertNotEqual(result.returncode, 0)
-            self.assertFalse((directory / 'launches.json').exists())
-            self.assertFalse((directory / 'control/ready.json').exists())
-
-
-@unittest.skipUnless(sys.platform=='win32', 'native Windows terminal required')
-class NativeTerminalTests(unittest.TestCase):
-    def setUp(self):
-        self.work=tempfile.TemporaryDirectory(prefix='movie-terminal-')
-        self.addCleanup(self.work.cleanup)
-        self.directory=Path(self.work.name)/'session'
-        self.shell=os.environ.get('MOVIE_TEST_SHELL','powershell51')
-        self.number=0
-        identity=subprocess.run(['whoami','/all'],capture_output=True,check=True)
-        self.assertIn(b'S-1-16-8192',identity.stdout)
-        (Path(self.work.name)/'identity.txt').write_bytes(identity.stdout)
-        self.log=(Path(self.work.name)/'serve.log').open('wb')
-        self.addCleanup(self.log.close)
-        argv=[sys.executable,str(SCRIPT),'serve','--shell',self.shell,'--directory',str(self.directory)]
-        for flag,variable in [('shell-exe','MOVIE_TEST_SHELL_EXE'),('ttyd','MOVIE_TEST_TTYD')]:
-            if os.environ.get(variable):argv+=['--'+flag,os.environ[variable]]
-        self.server=subprocess.Popen(argv,stdout=self.log,stderr=subprocess.STDOUT)
-        deadline=time.monotonic()+30
-        while not (self.directory/'control/ready.json').exists() and self.server.poll() is None and time.monotonic()<deadline:time.sleep(.05)
-        self.assertTrue((self.directory/'control/ready.json').exists(),(Path(self.work.name)/'serve.log').read_text(errors='replace'))
-
-    def tearDown(self):
-        if self.server.poll() is None:
-            try:self.request('cancel')
-            finally:
-                try:self.server.wait(10)
-                except subprocess.TimeoutExpired:self.server.kill();self.server.wait()
-        self.log.close()
-        evidence=os.environ.get('MOVIE_TERMINAL_EVIDENCE')
-        if evidence:
-            import shutil
-            destination=Path(evidence)/(self.shell+'-'+self._testMethodName)
-            shutil.copytree(self.work.name,destination,ignore=shutil.ignore_patterns('profile'),dirs_exist_ok=True)
-        self.work.cleanup()
-
-    def cli(self,*args):
-        return subprocess.run([sys.executable,str(SCRIPT),*args],capture_output=True,timeout=40)
+import fixtures
 
-    def request(self,operation,wait=True,expected=0,timeout=30,**fields):
-        self.number+=1
-        file=Path(self.work.name)/f'request-{self.number}.json'
-        file.write_text(json.dumps(dict(id=self.number,operation=operation,**fields)),encoding='utf-8')
-        args=['request','--directory',str(self.directory),'--file',str(file),'--timeout',str(timeout)]
-        if wait:args+=['--wait-result']
-        r=self.cli(*args)
-        self.assertEqual(r.returncode,expected,(r.stdout+r.stderr).decode(errors='replace'))
-        return json.loads(r.stdout) if r.stdout.strip() else None
+SCRIPT = Path(__file__).resolve().parents[2] / "skills/proving-it-works-with-a-movie/examples/film-terminal.py"
+FIXTURE = Path(__file__).resolve().with_name("fixtures") / "terminal_app.py"
+TTYD = os.environ.get("MOVIE_TEST_TTYD") or shutil.which("ttyd")
+BROWSER = fixtures.load_script("browser_tools").find_browser(os.environ.get("MOVIE_TEST_BROWSER"))
+SHELL = os.environ.get("MOVIE_TEST_SHELL") or ("powershell51" if os.name == "nt" else "bash")
+BASH = SHELL in ("bash", "gitbash")
 
-    def run_command(self,command,**kwargs):return self.request('run',command=command,**kwargs)
 
-    def native(self,code):
-        import base64
-        payload=base64.b64encode(code.encode()).decode()
-        exe=sys.executable.replace('\\','/')
-        return ('' if self.shell=='gitbash' else '& ')+f"'{exe}' -c \"import base64;exec(base64.b64decode('{payload}'))\""
+def recorder():
+    spec = importlib.util.spec_from_file_location("film_terminal", SCRIPT)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
 
-    def test_takes_pending_command_and_failure_recovery(self):
-        ready=json.loads((self.directory/'control/ready.json').read_text())
-        self.assertGreaterEqual(ready['geometry']['columns'],80)
-        self.run_command("export MOVIE_VALUE=kept" if self.shell=='gitbash' else "$global:MovieValue='kept'")
-        self.request('begin-take',name='first')
-        r=self.run_command(self.native('import time;time.sleep(2);print("long command done")'),wait=True,expected=1,timeout=.2)
-        self.assertEqual(r['outcome'],'client-timeout')
-        pending=self.number
-        self.run_command("echo must-not-run",expected=1)
-        inspect=self.request('inspect');self.assertEqual(inspect['pending']['id'],pending)
-        first=self.request('end-take');self.assertEqual(first['kind'],'frames')
-        self.request('begin-take',name='second')
-        r=self.cli('result','--directory',str(self.directory),'--id',str(pending),'--timeout','10')
-        self.assertEqual(r.returncode,0,r.stdout+r.stderr)
-        fixture=Path(__file__).parent/'fixtures/terminal_app.py'
-        code=f"import runpy,sys;sys.argv=['fixture','states'];runpy.run_path({str(fixture)!r},run_name='__main__')"
-        self.run_command(self.native(code),wait=False,native_producer=sys.executable)
-        tui=self.number
-        time.sleep(6.5)
-        self.request('key',key='q')
-        self.assertEqual(self.cli('result','--directory',str(self.directory),'--id',str(tui)).returncode,0)
-        take=self.request('end-take')
-        self.assertEqual(take['session_id'],ready['session_id'])
-        from PIL import Image
-        colors=[]
-        for png in sorted(Path(take['src']).glob('*.png')):
-            # Require a solid block; a colored shell prompt is not a TUI state.
-            image=Image.open(png).convert('RGB')
-            pixel=image.getpixel((100,60))
-            if sum(count for count,value in image.crop((40,30,440,140)).getcolors(44000) if value==pixel)<30000:continue
-            if pixel[0]>pixel[1]*1.5 and pixel[0]>pixel[2]*1.5:color='red'
-            elif pixel[1]>pixel[0]*1.5 and pixel[1]>pixel[2]*1.5:color='green'
-            elif pixel[2]>pixel[0]*1.5 and pixel[2]>pixel[1]*1.3:color='blue'
-            else:continue
-            if not colors or colors[-1]!=color:colors.append(color)
-        self.assertEqual(colors,['red','green','blue'])
-        command="test \"$MOVIE_VALUE\" = kept" if self.shell=='gitbash' else "if ($MovieValue -ne 'kept') {throw 'state lost'}"
-        self.run_command(command)
-        logging=' | cat' if self.shell=='gitbash' else ' | Tee-Object -Variable MovieLog'
-        failure=self.run_command(self.native('import sys;print("producer");sys.exit(7)')+logging,native_producer=sys.executable,expected=1)
-        self.assertEqual(failure['producer_exit_code'],7)
-        self.run_command('echo recovered')
-        self.request('close');self.assertEqual(self.server.wait(10),0)
 
-    def outcome_commands(self):
-        """Shell outcomes the recorder must attribute correctly: (name, command, native_producer)."""
-        exe=os.environ['MOVIE_TEST_SHELL_EXE'];python=sys.executable
-        commands=[('native_success',self.native('import sys;sys.exit(0)'),python)]
-        if self.shell!='gitbash':
-            commands+=[
-                ('cmdlet_failure',"Get-Item 'Z:\\probe-path-that-does-not-exist'",None),
-                ('native_failure',self.native('import sys;sys.exit(7)'),python),
-                ('cmdlet_success',"Write-Output 'cmdlet success λ'",None),
-                ('terminating_error',"throw 'probe terminating error'",None),
-                ('nonterminating_error',"Write-Error 'probe nonterminating error'",None),
-                ('parse_failure','Write-Output )',None),
-                ('logging_failure',self.native("import sys;print('producer');sys.exit(7)")+' | Tee-Object -Variable ProbeLog',python),
-                ('expression_wrapper',"(Write-Error 'probe expression wrapper')",None),
-                ('script_exit',f"& '{exe}' -NoLogo -NoProfile -Command 'exit 9'",exe),
-            ]
+def gone(pid, timeout=5):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if os.name == "nt":
+            listed = subprocess.run(["tasklist", "/FI", f"PID eq {pid}", "/NH"],
+                                    capture_output=True, text=True).stdout
+            if str(pid) not in listed:
+                return True
         else:
-            commands+=[
-                ('shell_failure','test -e /probe-path-that-does-not-exist',None),
-                ('native_failure',self.native('import sys;sys.exit(7)'),python),
-                ('shell_success',"printf 'shell success λ\\n'",None),
-                ('parse_failure','echo )',None),
-                ('logging_failure',self.native("import sys;print('producer');sys.exit(7)")+' | cat',python),
-                ('script_exit',"bash --noprofile --norc -c 'exit 9'",'Git Bash'),
-            ]
-        return commands
-
-    def test_native_shell_outcomes(self):
-        for name,command,native in self.outcome_commands():
-            success=name in ('native_success','cmdlet_success','shell_success')
-            result=self.run_command(command,native_producer=native,expected=0 if success else 1)
-            self.assertEqual(result['outcome'],'completed')
-            if name=='expression_wrapper':
-                self.assertEqual(result['raw_shell_success'],self.shell=='powershell51')
-            if name=='logging_failure':self.assertEqual(result['producer_exit_code'],7)
-        if self.shell != 'gitbash':
-            self.run_command(self.native('import sys;sys.exit(0)'),native_producer=sys.executable)
-            missing=self.run_command("Write-Output 'no native process ran'",native_producer=sys.executable,expected=1)
-            self.assertIsNone(missing['producer_exit_code'])
-        self.request('close')
-        self.assertEqual(self.server.wait(10),0)
-
-    def test_command_deadline_ends_session_and_marks_take_incomplete(self):
-        self.request('begin-take',name='timeout')
-        result=self.run_command(self.native('import time;time.sleep(30)'),timeout_seconds=.4,expected=1)
-        self.assertEqual(result['outcome'],'unknown')
-        self.assertIsNone(result['shell_success'])
-        self.assertNotEqual(self.server.wait(10),0)
-        self.assertTrue(json.loads((self.directory/'timeout/take.json').read_text())['incomplete'])
-
-    def test_gap_duplicate_and_rejection_leave_cancellation_usable(self):
-        file=Path(self.work.name)/'gap.json';file.write_text('{"id":2,"operation":"inspect"}')
-        result=self.cli('request','--directory',str(self.directory),'--file',str(file),'--timeout','1')
-        self.assertNotEqual(result.returncode,0)
-        self.assertFalse((self.directory/'control/000002.request.json').exists())
-        self.request('key',key='F99',expected=1)
-        self.assertEqual(json.loads((self.directory/'control/status.json').read_text())['next_request_id'],2)
-        first=Path(self.work.name)/'request-1.json'
-        result=self.cli('request','--directory',str(self.directory),'--file',str(first),'--timeout','1')
-        self.assertNotEqual(result.returncode,0)
-        self.request('cancel');self.assertEqual(self.server.wait(10),0)
-
-    def own_descendants(self):
-        fixture=Path(__file__).parent/'fixtures/terminal_app.py'
-        directory=Path(self.work.name)/'descendants'
-        code=f"import runpy,sys;sys.argv=['fixture','tree',{str(directory)!r}];runpy.run_path({str(fixture)!r},run_name='__main__')"
-        self.run_command(self.native(code),wait=False,native_producer=sys.executable)
-        deadline=time.monotonic()+10
-        while time.monotonic()<deadline and len(list(directory.glob('*.json')))<3:time.sleep(.05)
-        pids=[json.loads(p.read_text())['pid'] for p in directory.glob('*.json')]
-        self.assertEqual(len(pids),3)
-        launches=json.loads((self.directory/'launches.json').read_text())
-        return pids+[v['pid'] for v in launches]
-
-    def assert_dead(self,pids):
-        import ctypes
-        from ctypes import wintypes as W
-        kernel=ctypes.WinDLL('kernel32',use_last_error=True)
-        kernel.OpenProcess.argtypes=[W.DWORD,W.BOOL,W.DWORD];kernel.OpenProcess.restype=W.HANDLE
-        kernel.WaitForSingleObject.argtypes=[W.HANDLE,W.DWORD];kernel.CloseHandle.argtypes=[W.HANDLE]
-        deadline=time.monotonic()+10
-        alive=pids
-        while alive and time.monotonic()<deadline:
-            alive=[]
-            for pid in pids:
-                handle=kernel.OpenProcess(0x100000,False,pid)
-                if handle:
-                    if kernel.WaitForSingleObject(handle,0)==258:alive.append(pid)
-                    kernel.CloseHandle(handle)
-            if alive:time.sleep(.05)
-        self.assertEqual(alive,[])
-
-    def cleanup_case(self,mode):
-        sentinel=subprocess.Popen([sys.executable,'-c','import time;time.sleep(180)'])
-        try:
-            pids=self.own_descendants()
-            self.request('begin-take',name='cleanup')
-            if mode in ('close','cancel'):
-                self.request(mode);self.assertEqual(self.server.wait(10),0)
-                take=json.loads((self.directory/'cleanup/take.json').read_text())
-                self.assertEqual(take['incomplete'],mode=='cancel')
-            elif mode=='forced-exit':self.server.kill();self.server.wait(10)
-            elif mode=='browser-loss':
-                launches=json.loads((self.directory/'launches.json').read_text())
-                browser=next(v['pid'] for v in launches if v['name']=='browser')
-                subprocess.run(['taskkill','/PID',str(browser),'/F'],capture_output=True,check=True)
-                self.assertNotEqual(self.server.wait(10),0)
-            elif mode=='capture-write':
-                samples=self.directory/'cleanup/samples'
-                samples.rename(samples.with_name('saved-samples'));samples.write_text('blocked')
-                self.assertNotEqual(self.server.wait(10),0)
-            elif mode=='report-write':
-                result_path=self.directory/'control'/f'{self.number+1:06d}.result.json';result_path.mkdir()
-                self.request('close',expected=1,timeout=1)
-                self.assertNotEqual(self.server.wait(10),0)
-            elif mode=='control-write':
-                status=self.directory/'control/status.json';status.unlink();status.mkdir()
-                # Publish directly: client status validation correctly cannot read a directory.
-                number=self.number+1
-                (self.directory/'control'/f'{number:06d}.request.json').write_text(json.dumps(dict(id=number,operation='inspect')))
-                self.assertNotEqual(self.server.wait(10),0)
-            self.assert_dead(pids)
-            self.assertIsNone(sentinel.poll())
-            (Path(self.work.name)/'cleanup-evidence.json').write_text(json.dumps(dict(mode=mode,owned_pids=pids,owned_remaining=[],sentinel_pid=sentinel.pid,sentinel_alive=True)))
-        finally:
-            sentinel.terminate();sentinel.wait(10)
-
-    def test_normal_cleanup(self):self.cleanup_case('close')
-    def test_cancel_cleanup(self):self.cleanup_case('cancel')
-    def test_browser_loss_cleanup(self):self.cleanup_case('browser-loss')
-    def test_forced_exit_cleanup(self):self.cleanup_case('forced-exit')
-    def test_capture_write_failure_cleanup(self):self.cleanup_case('capture-write')
-    def test_report_write_failure_cleanup(self):self.cleanup_case('report-write')
-    def test_control_write_failure_cleanup(self):self.cleanup_case('control-write')
-
-    def test_geometry_change_ends_session(self):
-        pids=self.own_descendants()
-        self.request('begin-take',name='geometry')
-        launches=json.loads((self.directory/'launches.json').read_text())
-        browser=next(v for v in launches if v['name']=='browser')
-        port=next(v.split('=')[1] for v in browser['argv'] if v.startswith('--remote-debugging-port='))
-        import urllib.request,websocket
-        with urllib.request.urlopen(f'http://127.0.0.1:{port}/json/list') as response:pages=json.load(response)
-        page=next(page for page in pages if page['type']=='page' and page['url'].startswith('http://127.0.0.1:'))
-        ws=websocket.create_connection(page['webSocketDebuggerUrl'],suppress_origin=True)
-        try:
-            ws.send(json.dumps(dict(id=1,method='Runtime.evaluate',params=dict(expression="document.querySelector('.xterm').parentElement.style.width='800px'; window.dispatchEvent(new Event('resize')); document.querySelector('.xterm').parentElement.getBoundingClientRect().width",returnByValue=True))))
-            resize=json.loads(ws.recv())
-            self.assertNotIn('error',resize)
-            (Path(self.work.name)/'resize-response.json').write_text(json.dumps(dict(page=page,response=resize)))
-            self.assertEqual(resize['result']['result'].get('value'),800,resize)
-            self.assertNotEqual(self.server.wait(10),0)
-        finally:ws.close()
-        self.assert_dead(pids)
-        self.assertTrue(json.loads((self.directory/'geometry/take.json').read_text())['incomplete'])
-
-
-class CaptureBoundaryTests(unittest.TestCase):
-    recorder = TerminalPolicyTests.recorder
-    def test_screenshot_deadline_is_bounded(self):
-        from types import SimpleNamespace
-        m=self.recorder();recorder=m.Recorder.__new__(m.Recorder)
-        recorder.take={'samples':[]}
-        recorder.capture=(1,time.monotonic()-2.01)
-        recorder.terminal=SimpleNamespace(cdp=SimpleNamespace(responses={}))
-        with self.assertRaises(TimeoutError):recorder.capture_tick()
-
-    def test_end_take_drains_existing_screenshot_before_next_take(self):
-        from types import SimpleNamespace
-        m=self.recorder()
-        with tempfile.TemporaryDirectory() as temporary:
-            directory=Path(temporary);(directory/'samples').mkdir()
-            (directory/'samples/0.png').write_bytes(b'first')
-            recorder=m.Recorder.__new__(m.Recorder)
-            start=time.monotonic()
-            recorder.take=dict(name='one',directory=str(directory),start=start,begin_request=1,
-                               samples=[dict(path=str(directory/'samples/0.png'),requested=start,completed=start)])
-            recorder.capture=(9,start);recorder.capture_due=start+100
-            class CDP:
-                responses={}
-                def pump(self):self.responses[9]={'result':{'data':'c2Vjb25k'}}
-            recorder.geometry=dict(columns=220,rows=59)
-            recorder.terminal=SimpleNamespace(cdp=CDP(),closed=False,terminal_sizes=[recorder.geometry.copy()])
-            result=recorder.end()
-            self.assertEqual(len(result['samples']),2)
-            self.assertFalse(recorder.terminal.cdp.responses)
-
-    def test_late_screenshot_response_is_still_a_timeout(self):
-        from types import SimpleNamespace
-        m=self.recorder();recorder=m.Recorder.__new__(m.Recorder)
-        with tempfile.TemporaryDirectory() as temporary:
-            (Path(temporary)/'samples').mkdir()
-            recorder.take={'directory':temporary,'samples':[{}]}
-            recorder.capture_due=time.monotonic()+10
-            recorder.capture=(1,time.monotonic()-2.01)
-            recorder.terminal=SimpleNamespace(cdp=SimpleNamespace(responses={1:{'result':{'data':'YQ=='}}}))
-            with self.assertRaises(TimeoutError):recorder.capture_tick()
-
-class ClientWaitTests(unittest.TestCase):
-    recorder = TerminalPolicyTests.recorder
-
-    def test_ack_and_result_share_one_client_wait_budget(self):
-        import threading
-        from types import SimpleNamespace
-        m=self.recorder()
-        with tempfile.TemporaryDirectory() as temporary:
-            directory=Path(temporary);control=directory/'control';control.mkdir()
-            m.write_json(control/'status.json',dict(next_request_id=1,closed=False))
-            request=directory/'request.json';m.write_json(request,dict(id=1,operation='run',command='waiting'))
-            def acknowledge():
-                while not (control/'000001.request.json').exists():time.sleep(.01)
-                time.sleep(.4)
-                m.write_json(control/'000001.ack.json',dict(accepted=True))
-            worker=threading.Thread(target=acknowledge);worker.start()
-            started=time.monotonic()
-            try:reply,code=m.client(SimpleNamespace(directory=directory,action='request',file=request,timeout=.6,wait_result=True))
-            finally:worker.join()
-            self.assertEqual(code,1)
-            self.assertEqual(reply['outcome'],'client-timeout')
-            self.assertLess(time.monotonic()-started,.85)
-            before=sorted(p.name for p in control.iterdir())
-            m.write_json(control/'000001.result.json',dict(outcome='completed',success=True))
-            reply,code=m.client(SimpleNamespace(directory=directory,action='result',id=1,timeout=.1))
-            self.assertEqual(code,0)
-            self.assertEqual(sorted(p.name for p in control.iterdir()),sorted(before+['000001.result.json']))
-
-
-class FinalizationHealthTests(unittest.TestCase):
-    recorder = TerminalPolicyTests.recorder
-
-    def session(self, fault=None):
-        """Exercise real recorder/control files; inject CDP events at the drain boundary."""
-        from types import SimpleNamespace
-        from unittest.mock import patch
-        m=self.recorder()
-        temporary=tempfile.TemporaryDirectory()
-        self.addCleanup(temporary.cleanup)
-        directory=Path(temporary.name)/'session'
-
-        class Terminal:
-            def __init__(self,args,directory):
-                directory.mkdir()
-                self.session='finalization-session'
-                self.closed=False
-                self.released=False
-                self.terminal_sizes=[dict(columns=220,rows=59)]
-                self.parser=SimpleNamespace(records=[])
-                terminal=self
-                class CDP:
-                    def __init__(self):self.responses={};self.pumps=0
-                    def pump(self):
-                        self.pumps+=1
-                        if self.pumps==2:
-                            if fault=='loss':terminal.closed=True
-                            if fault=='geometry':terminal.terminal_sizes.append(dict(columns=100,rows=59))
-                            self.responses[9]={'result':{'data':'c2Vjb25k'}}
-                self.cdp=CDP()
-            def start(self):pass
-            def readiness(self):return dict(session=self.session,shell='powershell51',cwd=str(directory))
-            def install_prompt(self):pass
-            def close(self):self.released=True
-
-        with patch.object(m,'Terminal',Terminal):
-            recorder=m.Recorder(SimpleNamespace(directory=directory))
-        recorder.geometry=dict(columns=220,rows=59)
-        recorder.begin('take',0)
-        start=time.monotonic()
-        sample=directory/'take/samples/000000.png';sample.write_bytes(b'first')
-        recorder.take.update(start=start,samples=[dict(path=str(sample),requested=start,completed=start)])
-        recorder.capture=(9,start)
-        recorder.capture_due=start+100
-        return m,recorder
-
-    def test_close_rejects_terminal_loss_during_final_capture(self):
-        self.check_close_fault('loss',ConnectionError)
-
-    def test_close_rejects_geometry_change_during_final_capture(self):
-        self.check_close_fault('geometry',RuntimeError)
-
-    def check_close_fault(self,fault,error):
-        m,recorder=self.session(fault)
-        m.write_json(recorder.control/'000001.request.json',dict(id=1,operation='close'))
-        with self.assertRaises(error):recorder.run()
-        self.assertTrue(recorder.terminal.released)
-        self.assertTrue(m.read_json(recorder.args.directory/'take/take.json')['incomplete'])
-        self.assertFalse(m.read_json(recorder.control/'000001.result.json')['success'])
+            try:
+                os.kill(pid, 0)
+            except ProcessLookupError:
+                return True
+        time.sleep(0.1)
+    return False
+
+
+class PromptTests(unittest.TestCase):
+    def test_prompts_parse_both_terminators_and_paths_with_semicolons(self):
+        module = recorder()
+        log = (b"noise\x1b]0;MOVIE;1;1;;C:\\a;b\x07\x1b[0m"
+               b"\x1b]2;MOVIE;2;0;7;/c/x\x1b\\tail"
+               b"\x1b]0;MOVIE;3;1;0;/home/me\x07")
+        self.assertEqual(module.prompts(log), [
+            dict(n=1, ok=True, exit_code=None, cwd="C:\\a;b"),
+            dict(n=2, ok=False, exit_code=7, cwd="/c/x"),
+            dict(n=3, ok=True, exit_code=0, cwd="/home/me"),
+        ])
+        self.assertEqual(module.prompts(b"\x1b]0;something else\x07"), [])
+
+    def test_prompt_install_is_one_typed_line_per_shell(self):
+        module = recorder()
+        cwd = Path("C:/Users/x/movie O'Brien λ")
+        for kind in module.SHELLS:
+            line = module.prompt_command(kind, cwd)
+            self.assertEqual(len(line.splitlines()), 1, kind)
+            self.assertNotIn("MOVIE;", line, "the marker text must not be echoed by the install line")
+            self.assertIn("Brien λ", module.prompt_script(kind, cwd), "the script enters the cwd")
+
+    def test_keys_are_named_or_single_characters(self):
+        module = recorder()
+        self.assertEqual(module.key_params("Ctrl-C")["modifiers"], 2)
+        self.assertEqual(module.key_params("Enter")["text"], "\r")
+        self.assertEqual(module.key_params("q"), dict(key="q", text="q"))
+        with self.assertRaises(SystemExit):
+            module.key_params("Bogus")
+
+
+class FilmGridTests(unittest.TestCase):
+    def test_a_slow_capture_repeats_the_previous_frame_and_filming_holds_after_the_prompt(self):
+        module = recorder()
+        clock = {"now": 0.0}
+        shots = []
+
+        def capture():
+            shots.append(len(shots) + 1)
+            clock["now"] += 0.5 if len(shots) == 2 else 0.01  # the second screenshot stalls
+            return bytes([len(shots)])
+
+        with tempfile.TemporaryDirectory() as directory:
+            out = Path(directory) / "take"
+            frames = module.film(out, seconds=10, hold=0.4, capture=capture,
+                                 finished=lambda: clock["now"] >= 1.0,
+                                 clock=lambda: clock["now"],
+                                 sleep=lambda s: clock.__setitem__("now", clock["now"] + s))
+            files = sorted(out.glob("f*.png"))
+            self.assertEqual([f.name for f in files], [f"f{i:05d}.png" for i in range(frames)])
+            self.assertEqual(files[2].read_bytes(), files[1].read_bytes(), "missed slot repeats the last frame")
+            self.assertNotEqual(files[3].read_bytes(), files[2].read_bytes())
+            self.assertEqual(frames, 7, "1.0 s to the prompt plus 0.4 s hold at 5 fps")
+
+    def test_filming_stops_at_the_deadline_while_the_command_runs(self):
+        module = recorder()
+        clock = {"now": 0.0}
+        with tempfile.TemporaryDirectory() as directory:
+            frames = module.film(Path(directory), seconds=1.0, hold=5, capture=lambda: b"png",
+                                 finished=lambda: False, clock=lambda: clock["now"],
+                                 sleep=lambda s: clock.__setitem__("now", clock["now"] + s))
+            self.assertEqual(frames, 5)
+
+
+class ServeArgumentTests(unittest.TestCase):
+    def test_serve_refuses_a_missing_cwd_before_launching_anything(self):
+        with tempfile.TemporaryDirectory() as directory:
+            result = subprocess.run([sys.executable, str(SCRIPT), "serve", str(Path(directory) / "session"),
+                                     "--shell", "bash", "--cwd", str(Path(directory) / "missing")],
+                                    capture_output=True, text=True, timeout=60)
+            self.assertEqual(result.returncode, 1)
+            self.assertIn("--cwd is not a directory", result.stderr)
+            self.assertFalse((Path(directory) / "session" / "session.json").exists())
+
+
+@unittest.skipUnless(TTYD and BROWSER, "ttyd and a Chrome-family browser are required")
+class SessionTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="movie-terminal-", ignore_cleanup_errors=True)
+        self.addCleanup(self.tmp.cleanup)
+        self.work = Path(self.tmp.name) / "movie O'Brien λ"
+        self.work.mkdir()
+        self.session = Path(self.tmp.name) / "session"
+        self.log = (Path(self.tmp.name) / "serve.log").open("wb")
+        self.addCleanup(self.log.close)
+        argv = [sys.executable, str(SCRIPT), "serve", str(self.session), "--shell", SHELL,
+                "--cwd", str(self.work), "--ttyd", TTYD, "--browser", BROWSER]
+        if os.environ.get("MOVIE_TEST_SHELL_EXE"):
+            argv += ["--shell-exe", os.environ["MOVIE_TEST_SHELL_EXE"]]
+        self.serve = subprocess.Popen(argv, stdout=self.log, stderr=subprocess.STDOUT)
+        deadline = time.monotonic() + 45
+        while not (self.session / "ready.json").exists() and self.serve.poll() is None \
+                and time.monotonic() < deadline:
+            time.sleep(0.1)
+        if not (self.session / "ready.json").exists():
+            report = "".join(f"--- {name}\n" + path.read_text(errors="replace") if path.exists() else ""
+                             for name, path in (("serve.log", Path(self.tmp.name) / "serve.log"),
+                                                ("ttyd.log", self.session / "ttyd.log"),
+                                                ("browser.log", self.session / "browser.log")))
+            self.fail(report)
 
-    def test_healthy_close_still_finalizes_after_draining(self):
-        m,recorder=self.session()
-        m.write_json(recorder.control/'000001.request.json',dict(id=1,operation='close'))
-        recorder.run()
-        self.assertTrue(recorder.terminal.released)
-        self.assertFalse(m.read_json(recorder.args.directory/'take/take.json')['incomplete'])
-        self.assertTrue(m.read_json(recorder.control/'000001.result.json')['success'])
+    def tearDown(self):
+        if self.serve.poll() is None:
+            self.cli("close", str(self.session))
+            try:
+                self.serve.wait(15)
+            except subprocess.TimeoutExpired:
+                self.serve.kill()
+                self.serve.wait()
+        for pid in json.loads((self.session / "session.json").read_text(encoding="utf-8"))["pids"]:
+            self.assertTrue(gone(pid), f"pid {pid} survived close")
+
+    def cli(self, *args, timeout=120):
+        result = subprocess.run([sys.executable, str(SCRIPT), *args], capture_output=True, timeout=timeout)
+        return result.returncode, result.stdout.decode("utf-8", "replace"), result.stderr.decode("utf-8", "replace")
+
+    def run_command(self, command, *extra):
+        code, out, err = self.cli("run", str(self.session), command, *extra)
+        self.assertTrue(out.strip(), err)
+        return code, json.loads(out.strip().splitlines()[-1])
+
+    def quoted(self, *words):
+        if BASH:
+            return " ".join(shlex.quote(w.replace("\\", "/")) for w in words)
+        return "& " + " ".join("'" + w.replace("'", "''") + "'" for w in words)
+
+    def native(self, code):
+        return self.quoted(sys.executable) + f' -c "{code}"'
+
+    def test_commands_report_status_and_the_shell_persists_between_calls(self):
+        code, result = self.run_command("echo hello")
+        self.assertEqual((code, result["outcome"], result["ok"]), (0, "completed", True), result)
+        self.assertTrue(result["cwd"].endswith("movie O'Brien λ"), result["cwd"])
+        if BASH:
+            set_value, check_value, failing = "MOVIE_VALUE=kept", 'test "$MOVIE_VALUE" = kept', "false"
+        else:
+            set_value = "$global:MovieValue = 'kept'"
+            check_value = "if ($global:MovieValue -ne 'kept') { throw 'lost' }"
+            failing = "Get-Item 'Z:\\nowhere'"
+        self.assertEqual(self.run_command(set_value)[0], 0)
+        self.assertEqual(self.run_command(check_value)[0], 0, "state must survive separate calls")
+        code, result = self.run_command(failing)
+        self.assertEqual((code, result["ok"]), (1, False), result)
+        code, result = self.run_command(self.native("import sys; sys.exit(7)"))
+        self.assertEqual((code, result["ok"], result["exit_code"]), (1, False, 7), result)
+
+    def test_a_tui_is_filmed_across_two_takes_and_a_long_command_across_calls(self):
+        from PIL import Image
 
-    def test_finalization_without_pending_capture_checks_observation(self):
-        for fault,error in [('loss',ConnectionError),('geometry',RuntimeError)]:
-            with self.subTest(fault=fault):
-                m,recorder=self.session()
-                recorder.capture=None
-                if fault=='loss':recorder.terminal.closed=True
-                else:recorder.terminal.terminal_sizes.append(dict(columns=100,rows=59))
-                with self.assertRaises(error):recorder.end()
-                recorder.end(incomplete=True)
-                self.assertTrue(m.read_json(recorder.args.directory/'take/take.json')['incomplete'])
+        take_one, take_two, take_three = (self.work / name for name in ("take-one", "take-two", "take-three"))
+        code, result = self.run_command(self.quoted(sys.executable, str(FIXTURE)),
+                                        "--record", str(take_one), "--seconds", "7")
+        self.assertEqual((code, result["outcome"]), (2, "running"), result)
+        frames = sorted(take_one.glob("f*.png"))
+        self.assertGreaterEqual(len(frames), 30, "7 s at 5 fps")
+        seen = []
+        for frame in frames:
+            with Image.open(frame) as image:
+                r, g, b = image.convert("RGB").getpixel((300, 120))
+            color = ("red" if r > 150 and g < 100 and b < 100 else
+                     "green" if g > 120 and r < 100 and b < 140 else
+                     "blue" if b > 150 and r < 100 and g < 140 else None)
+            if color and (not seen or seen[-1] != color):
+                seen.append(color)
+        self.assertEqual(seen, ["red", "green", "blue"], "the three TUI states, in order, in the automatic frames")
+        code, out, err = self.cli("key", str(self.session), "q", "--record", str(take_two), "--seconds", "10")
+        result = json.loads(out.strip().splitlines()[-1])
+        self.assertEqual((code, result["outcome"], result["ok"]), (0, "completed", True), (result, err))
+        self.assertGreaterEqual(result["frames"], 7, "the exit plus the 1.5 s hold")
+        self.assertEqual(len(json.loads((self.work / "states.json").read_text())), 3)
+
+        code, result = self.run_command(self.native("import time; time.sleep(3)"), "--seconds", "1")
+        self.assertEqual(result["outcome"], "running")
+        code, out, err = self.cli("watch", str(self.session), "--record", str(take_three), "--seconds", "15")
+        result = json.loads(out.strip().splitlines()[-1])
+        self.assertEqual((code, result["outcome"], result["ok"]), (0, "completed", True), (result, err))
+        self.assertGreaterEqual(result["frames"], 10, "about 2 s of waiting plus the hold")
+        self.assertEqual(result["scene"], {"kind": "frames", "src": str(take_three.resolve()), "rate": 5})
+
+    def test_close_kills_the_shell_tree_and_spares_unrelated_processes(self):
+        sentinel = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(120)"])
+        self.addCleanup(sentinel.kill)
+        tree = self.work / "tree"
+        code, result = self.run_command(self.quoted(sys.executable, str(FIXTURE), "tree", str(tree)), "--seconds", "2")
+        self.assertEqual(result["outcome"], "running")
+        deadline = time.monotonic() + 20
+        while len(list(tree.glob("*.json"))) < 3 and time.monotonic() < deadline:
+            time.sleep(0.1)
+        pids = [json.loads(path.read_text())["pid"] for path in tree.glob("*.json")]
+        self.assertEqual(len(pids), 3)
+        code, out, err = self.cli("close", str(self.session))
+        self.assertEqual(code, 0, err)
+        self.assertEqual(self.serve.wait(15), 0)
+        for pid in pids:
+            self.assertTrue(gone(pid), f"descendant {pid} survived close")
+        self.assertIsNone(sentinel.poll(), "an unrelated process must survive")
+
+
+if __name__ == "__main__":
+    unittest.main()