Make harness startup and turns lighter

.github/workflows/ci.yml

@@ -51,6 +51,9 @@ jobs:
       - name: JSON live-control protocol test
         run: python3 scripts/test-json-controls.py zig-out/bin/graff
 
+      - name: Live JSON stream contains no raw stdout lines
+        run: python3 scripts/test-json-live.py zig-out/bin/graff
+
       - name: PTY spinner anti-stealth test
         run: python3 scripts/test-pty-spinner.py zig-out/bin/graff
 
@@ -66,6 +69,9 @@ jobs:
       - name: Codex WS/SSE transport and mid-turn compaction test
         run: python3 scripts/test-pty-codex-ws.py zig-out/bin/graff
 
+      - name: AI title and first turn run in parallel
+        run: python3 scripts/test-title-parallel.py zig-out/bin/graff
+
       - name: Stream stall/drop is not a user Esc (#134/#132)
         run: python3 scripts/test-stall-drop.py zig-out/bin/graff
 

benchmarks/README.md

@@ -22,7 +22,7 @@ task, the model, and the network.
 | `cost.py` | USD per task (each tool's own usage, [gateway prices](https://codegraff.com/docs/models)) | graff/deepseek-v4-pro **$0.022** vs Claude Code/Opus-4.8 **$0.51** vs Codex/gpt-5.5 **$0.42** |
 | `memory.py` | Peak RSS + CPU (`/usr/bin/time -l`) | graff **~25 MB** floor vs Node **~410 MB** vs Rust **~206 MB** |
 | `latency.py` | One-shot turn latency, same gpt-5.5 endpoint, concurrent pairs | graff **4.4 s** vs codex **8.9 s** (one-shot only) |
-| `memory_session.py` | RSS-vs-turn slope of ONE persistent `--json` session (the #124 leak detector; `memory.py` is one-shot peak RSS and structurally blind to per-turn growth) | flat slope after warmup = no per-turn leak |
+| `memory_session.py` | RSS-vs-turn slope plus session/scratch arena capacity for ONE persistent `--json` session (the #124 leak detector; `memory.py` is one-shot peak RSS and structurally blind to per-turn growth) | flat slope after warmup = no per-turn leak |
 
 ```sh
 python3 benchmarks/cost.py
@@ -31,6 +31,11 @@ python3 benchmarks/latency.py 8
 python3 benchmarks/memory_session.py 50   # N turns, one persistent session
 ```
 
+`memory_session.py` drives and samples one turn at a time so the child remains
+alive at every RSS sample. Set `GRAFF_BENCH_MODEL` to test a model other than
+`deepseek-v4-pro`; the script enables `GRAFF_MEM_DEBUG` itself and prints both
+arena capacities beside process RSS.
+
 Tasks live in `tasks.py` (edit to test other work).
 
 ## Honest caveats (please read)

benchmarks/memory_session.py

@@ -13,13 +13,16 @@
 
 Run:  CODEGRAFF_API_KEY=... python3 benchmarks/memory_session.py [N] [graff-path]
       N defaults to 16; graff-path defaults to `graff` on PATH.
+      GRAFF_BENCH_MODEL selects the model (default: deepseek-v4-pro).
 """
 import subprocess, json, os, sys
 
 os.chdir(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 N = int(sys.argv[1]) if len(sys.argv) > 1 else 16
 GRAFF = sys.argv[2] if len(sys.argv) > 2 else "graff"
+MODEL = os.environ.get("GRAFF_BENCH_MODEL", "deepseek-v4-pro")
+SLOPE_LIMIT_MB = 0.1
 
 # Short prompts that each stream a paragraph — many SSE events per turn (the
 # parse-garbage axis #124 is about) without slow multi-tool work, so the RSS
@@ -44,40 +47,53 @@ def rss_mb(pid):
 
 
 def main():
-    env = {**os.environ, "GRAFF_NO_TELEMETRY": "1", "GRAFF_FLEET": "off"}
-    p = subprocess.Popen([GRAFF, "--model", "deepseek-v4-pro", "--json"],
+    if N < 2:
+        raise SystemExit("N must be at least 2")
+    env = {**os.environ, "GRAFF_NO_TELEMETRY": "1", "GRAFF_FLEET": "off",
+           "GRAFF_MEM_DEBUG": "1"}
+    p = subprocess.Popen([GRAFF, "--model", MODEL, "--json"],
                          stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                          stderr=subprocess.DEVNULL, text=True, env=env, bufsize=1)
     samples = []
     try:
-        # Queue all turns up front (small JSON lines fit the pipe buffer), keep
-        # stdin open so the child stays alive for the final RSS sample, then read
-        # + sample at each turn boundary. graff processes them sequentially.
+        # Drive one turn at a time and keep stdin open through every sample.
+        # Closing it up front races the final sample against process exit, which
+        # can produce a bogus 0 MB reading and a wildly negative slope.
         for i in range(N):
             p.stdin.write(json.dumps({"type": "user", "text": PROMPTS[i % len(PROMPTS)]}) + "\n")
-        p.stdin.flush()
-        p.stdin.close()  # EOF after the queued turns so graff drains them all then exits
-        done = 0
-        while done < N:
-            line = p.stdout.readline()
-            if not line:
-                raise RuntimeError("graff exited early after %d/%d turns" % (done, N))
-            try:
-                ev = json.loads(line)
-            except Exception:
-                continue
-            if ev.get("type") == "error":
-                print("  (error: %s)" % ev.get("message"))
-            if ev.get("type") == "turn" and ev.get("complete"):
-                done += 1
-                r = rss_mb(p.pid)
-                samples.append((done, r))
-                print("  turn %2d: RSS = %7.1f MB" % (done, r))
+            p.stdin.flush()
+            arena = None
+            while True:
+                line = p.stdout.readline()
+                if not line:
+                    raise RuntimeError("graff exited early after %d/%d turns (exit %s)" %
+                                       (i, N, p.poll()))
+                try:
+                    ev = json.loads(line)
+                except Exception:
+                    continue
+                if ev.get("type") == "error":
+                    raise RuntimeError("turn %d failed: %s" % (i + 1, ev.get("message")))
+                if ev.get("type") == "mem":
+                    arena = (ev.get("session_arena_kb", 0), ev.get("scratch_arena_kb", 0))
+                if ev.get("type") == "turn" and ev.get("complete"):
+                    r = rss_mb(p.pid)
+                    if r <= 0:
+                        raise RuntimeError("could not sample RSS for live graff process %d" % p.pid)
+                    samples.append((i + 1, r))
+                    detail = ("  arenas: session=%d KB scratch=%d KB" % arena) if arena else ""
+                    print("  turn %2d: RSS = %7.1f MB%s" % (i + 1, r, detail))
+                    break
     finally:
-        try:
-            p.stdin.close(); p.terminate(); p.wait(timeout=5)
-        except Exception:
-            p.kill()
+        if p.stdin and not p.stdin.closed:
+            p.stdin.close()
+        if p.poll() is None:
+            p.terminate()
+            try:
+                p.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                p.kill()
+                p.wait()
 
     tail = samples[2:] if len(samples) > 4 else samples  # drop warmup
     if len(tail) >= 2:
@@ -87,7 +103,12 @@ def main():
         mx, my = sum(xs) / n, sum(ys) / n
         den = sum((x - mx) ** 2 for x in xs) or 1.0
         slope = sum((x - mx) * (y - my) for x, y in tail) / den
-        verdict = "FLAT — no per-turn leak" if abs(slope) < 0.1 else "GROWING — leak present"
+        if slope > SLOPE_LIMIT_MB:
+            verdict = "GROWING — investigate retained memory"
+        elif slope < -SLOPE_LIMIT_MB:
+            verdict = "SHRINKING — no per-turn leak"
+        else:
+            verdict = "FLAT — no per-turn leak"
         print("\nRSS growth over turns %d..%d: %+.3f MB/turn  (%s)"
               % (xs[0], xs[-1], slope, verdict))
         print("first-tail %.1f MB -> peak %.1f MB" % (ys[0], max(ys)))

scripts/test-json-live.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Live local-backend proof that --json stdout stays strict JSONL."""
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+
+from codex_ws_mock import CodexMock, REPLY_TEXT
+
+
+_arg = sys.argv[1] if len(sys.argv) > 1 else "graff"
+GRAFF = os.path.abspath(_arg) if os.sep in _arg else _arg
+
+
+def decode(line: str) -> dict | None:
+    if not line.strip():
+        return None
+    try:
+        event = json.loads(line)
+    except json.JSONDecodeError as exc:
+        raise AssertionError(f"non-JSON stdout line: {line!r}") from exc
+    if not isinstance(event, dict) or not isinstance(event.get("type"), str):
+        raise AssertionError(f"malformed JSON event: {event!r}")
+    return event
+
+
+def main() -> None:
+    mock = CodexMock()
+    port = mock.start()
+    try:
+        with tempfile.TemporaryDirectory(prefix="graff-json-live-") as tmp:
+            codex_home = os.path.join(tmp, "codex-home")
+            os.makedirs(codex_home)
+            with open(
+                os.path.join(codex_home, "auth.json"), "w", encoding="utf-8"
+            ) as fh:
+                json.dump(
+                    {
+                        "tokens": {
+                            "access_token": "json-live-mock",
+                            "account_id": "acct-json-live",
+                        }
+                    },
+                    fh,
+                )
+            harness_dir = os.path.join(tmp, ".harness")
+            os.makedirs(harness_dir)
+            with open(
+                os.path.join(harness_dir, "settings.json"),
+                "w",
+                encoding="utf-8",
+            ) as fh:
+                json.dump({"skills": {"codedbpro": False}}, fh)
+
+            env = os.environ.copy()
+            for name in tuple(env):
+                if name.startswith("GRAFF_") or name.startswith("CODEX_"):
+                    env.pop(name)
+            env.update(
+                {
+                    "HOME": tmp,
+                    "CODEX_HOME": codex_home,
+                    "GRAFF_CODEX_URL": (
+                        f"http://127.0.0.1:{port}/backend-api/codex/responses"
+                    ),
+                    "GRAFF_FLEET": "off",
+                    "GRAFF_NO_TELEMETRY": "1",
+                }
+            )
+            process = subprocess.Popen(
+                [GRAFF, "--model", "codex", "--json", "--no-telemetry"],
+                cwd=tmp,
+                env=env,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                bufsize=1,
+            )
+            events: list[dict] = []
+            assert process.stdin is not None
+            assert process.stdout is not None
+            try:
+                for turn in range(2):
+                    process.stdin.write(
+                        json.dumps({"type": "user", "text": f"live turn {turn}"})
+                        + "\n"
+                    )
+                    process.stdin.flush()
+                    while True:
+                        line = process.stdout.readline()
+                        if not line:
+                            stderr = process.stderr.read() if process.stderr else ""
+                            raise AssertionError(
+                                f"graff exited early ({process.poll()}): {stderr}"
+                            )
+                        event = decode(line)
+                        if event is None:
+                            continue
+                        events.append(event)
+                        if event.get("type") == "turn" and event.get("complete"):
+                            break
+                process.stdin.close()
+                for line in process.stdout:
+                    event = decode(line)
+                    if event is not None:
+                        events.append(event)
+                code = process.wait(timeout=5)
+                if code != 0:
+                    stderr = process.stderr.read() if process.stderr else ""
+                    raise AssertionError(f"graff exit={code}: {stderr}")
+            finally:
+                if process.poll() is None:
+                    process.terminate()
+                    process.wait(timeout=5)
+
+            text_events = [event for event in events if event["type"] == "text"]
+            turns = [event for event in events if event["type"] == "turn"]
+            if len(text_events) != 2 or any(
+                event.get("text") != REPLY_TEXT for event in text_events
+            ):
+                raise AssertionError(f"missing unstreamed text events: {text_events!r}")
+            if len(turns) != 2 or any(
+                event.get("text") != REPLY_TEXT or not event.get("complete")
+                for event in turns
+            ):
+                raise AssertionError(f"bad terminal turn events: {turns!r}")
+            if mock.ws_turns != 2:
+                raise AssertionError(f"expected two WS turns, got {mock.ws_turns}")
+    finally:
+        mock.stop()
+    print("ok    live Codex WS emits strict JSONL, including fallback text events")
+
+
+if __name__ == "__main__":
+    main()