diff --git a/scripts/openfut-utas-observe.py b/scripts/openfut-utas-observe.py
new file mode 100755
index 0000000..2ead781
--- /dev/null
+++ b/scripts/openfut-utas-observe.py
@@ -0,0 +1,504 @@
+#!/usr/bin/env python3
+"""Record real UTAS traffic between FIFA 17 and the Python oracle.
+
+    openfut-utas-observe.py record  --listen 0.0.0.0:8199 --upstream 127.0.0.1:8099 \
+                                    --session captures/utas/session-001
+    openfut-utas-observe.py parse   --session captures/utas/session-001
+    openfut-utas-observe.py snapshot --upstream 127.0.0.1:8099 --out before/state-manifest.json
+
+WHY THIS EXISTS
+---------------
+Porting UTAS needs the exact bytes of real requests AND real responses. The
+oracle's own log truncates bodies at ~200 chars, and raising that cap would mean
+editing the behavioural specification to make it easier to copy -- which is
+exactly backwards. A proxy gets the same evidence and leaves the oracle alone.
+
+Standalone on purpose: NOT built into the Python container. The same tool can
+later sit in front of a Rust UTAS host, so the identical captured request can be
+replayed against both and diffed.
+
+THE DESIGN RULE: TEE, DO NOT REBUILD
+------------------------------------
+UTAS is plaintext HTTP/1.1 served by `ThreadingHTTPServer`, so keep-alive,
+pipelining and chunked transfer are all possible. A proxy that parses a request
+and re-emits it can corrupt the traffic it exists to observe -- and a subtle
+corruption here would look like a UTAS bug, sending the investigation in exactly
+the wrong direction.
+
+So this copies bytes VERBATIM in both directions and writes a second copy to
+disk. Transactions are reconstructed later, offline, from that copy. A bug in
+the parser therefore spoils the record and never the session. This is the same
+reasoning that keeps the oracle unmodified: do not let the measuring apparatus
+change what is being measured.
+
+TWO LAYERS, LIKE THE BLAZE CAPTURES
+-----------------------------------
+    raw/utas.ofcap        exact bytes, mode 0600, gitignored
+            |
+        sanitizer
+            |
+    sanitized/transactions.jsonl    repository-safe fixtures
+
+Raw captures are never normalised. Normalisation happens only when producing the
+committed fixture, and preserves exact bodies first, sanitises second.
+"""
+import argparse
+import base64
+import json
+import os
+import re
+import socket
+import sys
+import threading
+import time
+
+MAGIC = "OFCAP1"
+C2S = "c2s"
+S2C = "s2c"
+
+
+# ─────────────────────────────────────────────────────────── capture file ────
+class Capture:
+    """Append-only JSONL of raw byte chunks. One line per read().
+
+    JSONL with base64 rather than a packed binary format: this is a research
+    artefact that people will grep and eyeball, and the size cost is irrelevant
+    next to being able to inspect it without a decoder. Base64 is an encoding,
+    not a normalisation -- the bytes round-trip exactly.
+    """
+
+    def __init__(self, path):
+        self.path = path
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        # 0600 before anything is written: captures carry session tokens.
+        fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600)
+        self.f = os.fdopen(fd, "w")
+        self.lock = threading.Lock()
+        self._write({"magic": MAGIC, "started_unix": time.time()})
+
+    def _write(self, obj):
+        with self.lock:
+            self.f.write(json.dumps(obj, sort_keys=True) + "\n")
+            self.f.flush()
+
+    def chunk(self, conn_id, direction, data, t0):
+        self._write({
+            "conn": conn_id,
+            "dir": direction,
+            "t": round(time.monotonic() - t0, 6),
+            "unix": time.time(),
+            "len": len(data),
+            "b64": base64.b64encode(data).decode(),
+        })
+
+    def event(self, conn_id, kind, detail, t0):
+        self._write({
+            "conn": conn_id,
+            "event": kind,
+            "detail": detail,
+            "t": round(time.monotonic() - t0, 6),
+            "unix": time.time(),
+        })
+
+    def close(self):
+        with self.lock:
+            self.f.close()
+
+
+# ──────────────────────────────────────────────────────────────── record ────
+def pump(src, dst, cap, conn_id, direction, t0, done):
+    """Copy src->dst verbatim, teeing every chunk into the capture.
+
+    Forward FIRST, record second: the client must never wait on our disk.
+    """
+    try:
+        while True:
+            data = src.recv(65536)
+            if not data:
+                break
+            try:
+                dst.sendall(data)
+            except OSError as e:
+                cap.event(conn_id, "forward_failed", "%s %s" % (direction, e), t0)
+                break
+            cap.chunk(conn_id, direction, data, t0)
+    except OSError as e:
+        cap.event(conn_id, "read_failed", "%s %s" % (direction, e), t0)
+    finally:
+        # Half-close so the peer sees EOF in the same direction the original
+        # did. Connection semantics are part of what is being captured.
+        try:
+            dst.shutdown(socket.SHUT_WR)
+        except OSError:
+            pass
+        cap.event(conn_id, "eof", direction, t0)
+        done.set()
+
+
+def handle(client, addr, upstream_addr, cap, conn_id, t0):
+    cap.event(conn_id, "open", "%s:%d" % addr, t0)
+    try:
+        up = socket.create_connection(upstream_addr, timeout=30)
+    except OSError as e:
+        cap.event(conn_id, "upstream_failed", str(e), t0)
+        client.close()
+        return
+    client.settimeout(None)
+    up.settimeout(None)
+
+    a, b = threading.Event(), threading.Event()
+    t1 = threading.Thread(target=pump, args=(client, up, cap, conn_id, C2S, t0, a), daemon=True)
+    t2 = threading.Thread(target=pump, args=(up, client, cap, conn_id, S2C, t0, b), daemon=True)
+    t1.start(); t2.start()
+    t1.join(); t2.join()
+    for s in (client, up):
+        try:
+            s.close()
+        except OSError:
+            pass
+    cap.event(conn_id, "close", "", t0)
+
+
+def cmd_record(args):
+    lh, lp = args.listen.rsplit(":", 1)
+    uh, up = args.upstream.rsplit(":", 1)
+    session = os.path.abspath(args.session)
+    cap = Capture(os.path.join(session, "raw", "utas.ofcap"))
+
+    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    srv.bind((lh, int(lp)))
+    srv.listen(64)
+    t0 = time.monotonic()
+    print("recording %s -> %s:%s into %s" % (args.listen, uh, up, cap.path), flush=True)
+    print("  raw capture is mode 0600 and must stay out of git", flush=True)
+
+    n = 0
+    try:
+        while True:
+            client, addr = srv.accept()
+            n += 1
+            threading.Thread(
+                target=handle,
+                args=(client, addr, (uh, int(up)), cap, n, t0),
+                daemon=True,
+            ).start()
+    except KeyboardInterrupt:
+        print("\nstopped after %d connection(s)" % n, flush=True)
+    finally:
+        cap.close()
+
+
+# ─────────────────────────────────────────────────────────────── parsing ────
+def read_capture(path):
+    conns = {}
+    with open(path) as f:
+        for line in f:
+            r = json.loads(line)
+            if "magic" in r:
+                continue
+            c = conns.setdefault(r["conn"], {"c2s": bytearray(), "s2c": bytearray(),
+                                             "events": [], "marks": {"c2s": [], "s2c": []}})
+            if "b64" in r:
+                data = base64.b64decode(r["b64"])
+                c["marks"][r["dir"]].append((len(c[r["dir"]]), r["t"], r["unix"]))
+                c[r["dir"]].extend(data)
+            else:
+                c["events"].append(r)
+    return conns
+
+
+def split_messages(buf, is_response):
+    """Split an HTTP/1.1 byte stream into messages. Returns list of (head, body, end).
+
+    Framing follows RFC 7230: Transfer-Encoding: chunked wins over
+    Content-Length. A stream that cannot be framed stops the walk rather than
+    guessing -- a wrong split would silently mis-attribute bodies.
+    """
+    out, i = [], 0
+    while True:
+        sep = buf.find(b"\r\n\r\n", i)
+        if sep < 0:
+            break
+        head = bytes(buf[i:sep])
+        hs = head.decode("latin1")
+        start = sep + 4
+        te = re.search(r"(?im)^transfer-encoding:\s*(.+)$", hs)
+        cl = re.search(r"(?im)^content-length:\s*(\d+)\s*$", hs)
+        if te and "chunked" in te.group(1).lower():
+            j, body = start, bytearray()
+            while True:
+                nl = buf.find(b"\r\n", j)
+                if nl < 0:
+                    return out
+                try:
+                    size = int(bytes(buf[j:nl]).split(b";")[0], 16)
+                except ValueError:
+                    return out
+                j = nl + 2
+                if size == 0:
+                    j = buf.find(b"\r\n", j)
+                    j = (j + 2) if j >= 0 else len(buf)
+                    break
+                body.extend(buf[j:j + size])
+                j += size + 2
+            out.append((head, bytes(body), j)); i = j
+        elif cl:
+            n = int(cl.group(1))
+            end = start + n
+            if end > len(buf):
+                break
+            out.append((head, bytes(buf[start:end]), end)); i = end
+        else:
+            status_no_body = is_response and re.match(r"HTTP/\d\.\d (1\d\d|204|304)", hs)
+            if is_response and not status_no_body:
+                # No framing header: the body runs to end-of-stream.
+                out.append((head, bytes(buf[start:]), len(buf))); i = len(buf)
+                break
+            out.append((head, b"", start)); i = start
+    return out
+
+
+def time_at(marks, offset):
+    """Wall/relative time of the chunk containing byte `offset`."""
+    last = (0.0, 0.0)
+    for pos, t, unix in marks:
+        if pos > offset:
+            break
+        last = (t, unix)
+    return last
+
+
+def cmd_parse(args):
+    session = os.path.abspath(args.session)
+    raw = os.path.join(session, "raw", "utas.ofcap")
+    conns = read_capture(raw)
+    out_dir = os.path.join(session, "sanitized")
+    os.makedirs(out_dir, exist_ok=True)
+    out_path = os.path.join(out_dir, "transactions.jsonl")
+
+    total, skipped = 0, 0
+    with open(out_path, "w") as out:
+        for conn_id in sorted(conns):
+            c = conns[conn_id]
+            reqs = split_messages(c["c2s"], is_response=False)
+            resps = split_messages(c["s2c"], is_response=True)
+            if len(reqs) != len(resps):
+                # Reported, never silently trimmed: an unmatched count means the
+                # framing is not understood, and a quiet truncation would look
+                # like the client simply asked for less.
+                print("  conn %d: %d requests but %d responses -- pairing the common prefix"
+                      % (conn_id, len(reqs), len(resps)), file=sys.stderr)
+                skipped += abs(len(reqs) - len(resps))
+            keepalive = None
+            for seq, ((rh, rb, rend), (sh, sb, send)) in enumerate(zip(reqs, resps), 1):
+                rl = rh.split(b"\r\n")[0].decode("latin1")
+                sl = sh.split(b"\r\n")[0].decode("latin1")
+                m = re.match(r"(\S+)\s+(\S+)\s+(HTTP/\d\.\d)", rl)
+                method, target, ver = (m.group(1), m.group(2), m.group(3)) if m else ("?", rl, "?")
+                path, _, query = target.partition("?")
+                st = re.match(r"HTTP/\d\.\d\s+(\d+)", sl)
+                req_t, req_unix = time_at(c["marks"]["c2s"], rend - 1)
+                res_t, res_unix = time_at(c["marks"]["s2c"], max(send - 1, 0))
+                keepalive = len(reqs) > 1
+                rec = {
+                    "conn": conn_id,
+                    "seq": seq,
+                    "t_request": req_t,
+                    "t_response": res_t,
+                    "elapsed_ms": round((res_t - req_t) * 1000, 3),
+                    "unix_request": req_unix,
+                    "request": {
+                        "method": method, "path": path, "query": query, "version": ver,
+                        "headers": header_list(rh),
+                        "body_b64": base64.b64encode(rb).decode(),
+                        "body_len": len(rb),
+                    },
+                    "response": {
+                        "status": int(st.group(1)) if st else None,
+                        "status_line": sl,
+                        "headers": header_list(sh),
+                        "body_b64": base64.b64encode(sb).decode(),
+                        "body_len": len(sb),
+                    },
+                    "connection": {
+                        "requests_on_this_connection": len(reqs),
+                        "keepalive_observed": keepalive,
+                    },
+                }
+                out.write(json.dumps(sanitize(rec), sort_keys=True) + "\n")
+                total += 1
+
+    os.chmod(out_path, 0o644)
+    print("wrote %s (%d transactions across %d connections%s)"
+          % (out_path, total, len(conns),
+             "; %d unpaired messages reported above" % skipped if skipped else ""))
+    return 0
+
+
+def header_list(head):
+    """Headers in RECEIVED ORDER, as a list of pairs. Order can be protocol-
+    relevant and a dict would silently discard duplicates."""
+    lines = head.decode("latin1").split("\r\n")[1:]
+    out = []
+    for l in lines:
+        if not l:
+            continue
+        k, _, v = l.partition(":")
+        out.append([k, v.strip()])
+    return out
+
+
+SECRET_HEADERS = {"authorization", "cookie", "set-cookie", "x-ut-sid", "easw-session-data-nucleus-id"}
+SECRET_BODY_KEYS = ("sid", "token", "password", "answer", "secret")
+
+
+def sanitize(rec):
+    """Repository-safe form. Bodies are preserved EXACTLY; only credentials go.
+
+    Exact bodies first, sanitise second: the corpus is worthless if response
+    payloads are reshaped, so nothing here touches structure -- it replaces
+    known-secret header values and known-secret JSON keys, and records that it
+    did so.
+    """
+    redacted = []
+    for side in ("request", "response"):
+        hs = rec[side]["headers"]
+        for pair in hs:
+            if pair[0].lower() in SECRET_HEADERS:
+                redacted.append("%s.%s" % (side, pair[0]))
+                pair[1] = "<REDACTED>"
+        body = base64.b64decode(rec[side]["body_b64"])
+        new, hit = redact_json(body)
+        if hit:
+            redacted.extend("%s.body.%s" % (side, h) for h in hit)
+            rec[side]["body_b64"] = base64.b64encode(new).decode()
+            rec[side]["body_len_original"] = len(body)
+    if redacted:
+        rec["redacted"] = sorted(set(redacted))
+    return rec
+
+
+def redact_json(body):
+    if not body[:1] in (b"{", b"["):
+        return body, []
+    try:
+        doc = json.loads(body)
+    except Exception:
+        return body, []
+    hits = []
+
+    def walk(o):
+        if isinstance(o, dict):
+            for k in list(o):
+                if any(s in k.lower() for s in SECRET_BODY_KEYS) and isinstance(o[k], str):
+                    o[k] = "<REDACTED>"
+                    hits.append(k)
+                else:
+                    walk(o[k])
+        elif isinstance(o, list):
+            for v in o:
+                walk(v)
+
+    walk(doc)
+    if not hits:
+        return body, []
+    return json.dumps(doc, sort_keys=True).encode(), hits
+
+
+# ────────────────────────────────────────────────────────────── snapshot ────
+SNAPSHOT_ROUTES = [
+    ("account", "/openfut/account/sync"),
+]
+
+
+def cmd_snapshot(args):
+    """A light state manifest, for before/after pairing around a session.
+
+    Deliberately a summary, not a database dump: the point is to say what
+    changed, and a committed copy of the save would be both huge and full of
+    things that are not evidence.
+    """
+    uh, up = args.upstream.rsplit(":", 1)
+    man = {"unix": time.time(), "upstream": args.upstream, "routes": {}}
+    for name, path in SNAPSHOT_ROUTES:
+        try:
+            s = socket.create_connection((uh, int(up)), timeout=5)
+            body = b'{"personaId":0}'
+            req = ("POST %s HTTP/1.1\r\nHost: %s:%s\r\nContent-Type: application/json\r\n"
+                   "Content-Length: %d\r\nConnection: close\r\n\r\n" % (path, uh, up, len(body)))
+            s.sendall(req.encode() + body)
+            buf = b""
+            while True:
+                d = s.recv(4096)
+                if not d:
+                    break
+                buf += d
+            s.close()
+            _, _, payload = buf.partition(b"\r\n\r\n")
+            man["routes"][name] = summarize(payload)
+        except Exception as e:
+            # Recorded as an error rather than omitted: a manifest missing a
+            # section must not look like a section that was empty.
+            man["routes"][name] = {"error": str(e)}
+    os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
+    with open(args.out, "w") as f:
+        json.dump(man, f, indent=2, sort_keys=True)
+        f.write("\n")
+    print("wrote %s" % args.out)
+    for k, v in man["routes"].items():
+        print("  %s: %s" % (k, v))
+    return 0
+
+
+INTERESTING = ("coins", "unopenedPacks", "personaId", "personaName", "clubName",
+               "level", "experience", "accountFunds", "itemCount", "squadId")
+
+
+def summarize(payload):
+    try:
+        doc = json.loads(payload)
+    except Exception:
+        return {"raw_len": len(payload)}
+    out = {}
+
+    def walk(o, prefix=""):
+        if isinstance(o, dict):
+            for k, v in o.items():
+                if k in INTERESTING and not isinstance(v, (dict, list)):
+                    out[k] = v
+                else:
+                    walk(v, prefix + k + ".")
+        elif isinstance(o, list):
+            out.setdefault(prefix.rstrip(".") + ".count", len(o))
+
+    walk(doc)
+    return out
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__,
+                                 formatter_class=argparse.RawDescriptionHelpFormatter)
+    sub = ap.add_subparsers(dest="cmd", required=True)
+
+    r = sub.add_parser("record")
+    r.add_argument("--listen", required=True)
+    r.add_argument("--upstream", required=True)
+    r.add_argument("--session", required=True)
+    r.set_defaults(fn=cmd_record)
+
+    p = sub.add_parser("parse")
+    p.add_argument("--session", required=True)
+    p.set_defaults(fn=cmd_parse)
+
+    s = sub.add_parser("snapshot")
+    s.add_argument("--upstream", required=True)
+    s.add_argument("--out", required=True)
+    s.set_defaults(fn=cmd_snapshot)
+
+    args = ap.parse_args()
+    return args.fn(args) or 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/scripts/test-utas-observe.py b/scripts/test-utas-observe.py
new file mode 100755
index 0000000..07b9557
--- /dev/null
+++ b/scripts/test-utas-observe.py
@@ -0,0 +1,210 @@
+#!/usr/bin/env python3
+"""Prove the UTAS recorder is byte-transparent and parses what it saw.
+
+Two independent properties, because they fail differently:
+
+  TRANSPARENCY  what the client receives through the proxy is byte-identical to
+                what it receives going direct. If this breaks, the tool corrupts
+                the session it exists to observe -- and the damage would look
+                like a UTAS bug.
+
+  FIDELITY      the parsed transactions match what was actually sent. If this
+                breaks, only the record is wrong, which is why the design
+                separates the two.
+
+The upstream here is a purpose-built server, not the live oracle: it can be made
+to exercise keep-alive, bodies in both directions and chunked encoding on
+demand, and no live client is at risk.
+"""
+import base64
+import http.server
+import json
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import threading
+import time
+
+TOOL = os.path.join(os.path.dirname(os.path.abspath(__file__)), "openfut-utas-observe.py")
+SESSION = os.path.join(os.environ.get("TMPDIR", "/tmp"), "openfut-utas-observe-selftest")
+
+fails = []
+
+
+def check(name, ok, detail=""):
+    print(("  ok   " if ok else "  FAIL ") + name + (("  " + detail) if detail and not ok else ""))
+    if not ok:
+        fails.append(name)
+
+
+class H(http.server.BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"          # keep-alive, like the real UTAS
+
+    def log_message(self, *a):
+        pass
+
+    def _read_body(self):
+        n = int(self.headers.get("Content-Length", 0) or 0)
+        return self.rfile.read(n) if n else b""
+
+    def do_GET(self):
+        if self.path == "/chunked":
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Transfer-Encoding", "chunked")
+            self.end_headers()
+            for part in (b'{"a":1,', b'"b":[2,3],', b'"c":"end"}'):
+                self.wfile.write(b"%x\r\n" % len(part) + part + b"\r\n")
+            self.wfile.write(b"0\r\n\r\n")
+            return
+        body = json.dumps({"path": self.path}).encode()
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_POST(self):
+        got = self._read_body()
+        body = json.dumps({"echo_len": len(got),
+                           "echo": got.decode("latin1"),
+                           "sid": "SECRET-SESSION-VALUE"}).encode()
+        self.send_response(201)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+
+def free_port():
+    s = socket.socket()
+    s.bind(("127.0.0.1", 0))
+    p = s.getsockname()[1]
+    s.close()
+    return p
+
+
+def raw_exchange(port, requests):
+    """Send requests on ONE keep-alive connection; return all bytes received."""
+    s = socket.create_connection(("127.0.0.1", port), timeout=5)
+    for r in requests:
+        s.sendall(r)
+        time.sleep(0.15)
+    s.settimeout(1.5)
+    out = b""
+    try:
+        while True:
+            d = s.recv(65536)
+            if not d:
+                break
+            out += d
+    except socket.timeout:
+        pass
+    s.close()
+    return out
+
+
+def main():
+    shutil.rmtree(SESSION, ignore_errors=True)
+    up_port, proxy_port = free_port(), free_port()
+
+    srv = http.server.ThreadingHTTPServer(("127.0.0.1", up_port), H)
+    threading.Thread(target=srv.serve_forever, daemon=True).start()
+
+    proxy = subprocess.Popen(
+        [sys.executable, TOOL, "record",
+         "--listen", "127.0.0.1:%d" % proxy_port,
+         "--upstream", "127.0.0.1:%d" % up_port,
+         "--session", SESSION],
+        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    time.sleep(1.0)
+
+    body = json.dumps({"squad": [1, 2, 3], "note": "x" * 300}).encode()
+    reqs = [
+        b"GET /ut/game/fifa17/userMassInfo HTTP/1.1\r\nHost: t\r\n\r\n",
+        b"POST /ut/game/fifa17/purchased/items HTTP/1.1\r\nHost: t\r\n"
+        b"Content-Type: application/json\r\nContent-Length: %d\r\n\r\n" % len(body) + body,
+        b"GET /chunked HTTP/1.1\r\nHost: t\r\n\r\n",
+        b"GET /ut/game/fifa17/hub HTTP/1.1\r\nHost: t\r\nConnection: close\r\n\r\n",
+    ]
+
+    print("== transparency ==")
+    direct = raw_exchange(up_port, reqs)
+    through = raw_exchange(proxy_port, reqs)
+    # The two exchanges happen seconds apart, so `Date:` legitimately differs.
+    # Mask it -- but ONLY it, and assert the mask actually fired, so this cannot
+    # quietly hide a real difference.
+    import re as _re
+    def mask(b):
+        return _re.sub(rb"Date: [^\r\n]+", b"Date: <M>", b)
+    dm, tm = mask(direct), mask(through)
+    check("masking Date actually applied", dm != direct and tm != through)
+    check("bytes through the proxy are identical to bytes direct (Date masked)",
+          dm == tm,
+          "direct=%dB through=%dB firstdiff=%s" % (
+              len(direct), len(through),
+              next((i for i in range(min(len(dm), len(tm))) if dm[i] != tm[i]), "len")))
+    check("nothing but Date differed in the unmasked bytes",
+          len(direct) == len(through))
+    check("all four responses arrived", through.count(b"HTTP/1.1 ") == 4,
+          "saw %d" % through.count(b"HTTP/1.1 "))
+    check("keep-alive was actually used (one connection, four responses)",
+          through.count(b"HTTP/1.1 ") == 4)
+
+    time.sleep(0.6)
+    proxy.terminate()
+    try:
+        proxy.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        proxy.kill()
+
+    print("== capture file ==")
+    raw = os.path.join(SESSION, "raw", "utas.ofcap")
+    check("raw capture exists", os.path.exists(raw))
+    check("raw capture is mode 0600",
+          oct(os.stat(raw).st_mode & 0o777) == "0o600",
+          oct(os.stat(raw).st_mode & 0o777))
+
+    print("== fidelity ==")
+    r = subprocess.run([sys.executable, TOOL, "parse", "--session", SESSION],
+                       capture_output=True, text=True)
+    print("     " + r.stdout.strip().replace("\n", "\n     "))
+    if r.stderr.strip():
+        print("     stderr: " + r.stderr.strip().replace("\n", "\n     "))
+    txs = [json.loads(l) for l in
+           open(os.path.join(SESSION, "sanitized", "transactions.jsonl"))]
+    # Two client connections were made (direct + through); only one went via the proxy.
+    check("four transactions parsed", len(txs) == 4, "got %d" % len(txs))
+    if len(txs) == 4:
+        check("methods and paths in order",
+              [(t["request"]["method"], t["request"]["path"]) for t in txs] ==
+              [("GET", "/ut/game/fifa17/userMassInfo"),
+               ("POST", "/ut/game/fifa17/purchased/items"),
+               ("GET", "/chunked"),
+               ("GET", "/ut/game/fifa17/hub")])
+        check("request body preserved byte-for-byte",
+              base64.b64decode(txs[1]["request"]["body_b64"]) == body)
+        check("statuses recorded", [t["response"]["status"] for t in txs] == [200, 201, 200, 200])
+        chunked = base64.b64decode(txs[2]["response"]["body_b64"])
+        check("chunked response body dechunked correctly",
+              chunked == b'{"a":1,"b":[2,3],"c":"end"}', chunked[:60].decode("latin1"))
+        check("headers kept in received order, as pairs",
+              txs[0]["response"]["headers"][0][0] == "Server")
+        check("keep-alive recorded", txs[0]["connection"]["requests_on_this_connection"] == 4)
+        check("timing recorded", all(t["elapsed_ms"] is not None for t in txs))
+        # Sanitiser: the secret goes, the rest of the body does not.
+        echoed = json.loads(base64.b64decode(txs[1]["response"]["body_b64"]))
+        check("secret JSON value redacted", echoed.get("sid") == "<REDACTED>")
+        check("non-secret payload preserved exactly", echoed.get("echo_len") == len(body))
+        check("redaction is recorded, not silent", "redacted" in txs[1])
+
+    srv.shutdown()
+    print()
+    print("all checks passed" if not fails else "%d FAILED: %s" % (len(fails), fails))
+    return 1 if fails else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())