Update to Codex Replay

Changed files

Choose a file to see its text changes. Binary or large files show metadata only.

.codex-plugin/plugin.json →

OWNERS →

README.md →

TEMPEST.md →

assets/model-pricing.json →

docs/security-review/managed-network-settings.json →

docs/security-review/real-model-results.json →

integrity.json →

manual_tests/README.md →

manual_tests/egress_probe.py →

manual_tests/prompt_guard.py →

manual_tests/runtime_errors.py →

manual_tests/worker_isolation.py →

mcp/__pycache__/controller_constants.cpython-312.pyc →

mcp/__pycache__/final_results_receipt.cpython-312.pyc →

mcp/__pycache__/plugin_installations.cpython-312.pyc →

mcp/__pycache__/replay_batch.cpython-312.pyc →

mcp/__pycache__/replay_configuration.cpython-312.pyc →

mcp/__pycache__/server.cpython-312.pyc →

mcp/codex-worker.mjs →

mcp/controller-ranges.js →

mcp/controller.css →

mcp/controller.html →

mcp/replay_batch.py →

mcp/server.py →

mcp/worker-policy.json →

plans/MULTI_THREAD_REPLAY.md →

scripts/__pycache__/claude_code_sample_loader.cpython-312.pyc →

scripts/__pycache__/historical_discovery.cpython-312.pyc →

scripts/__pycache__/historical_discovery_capabilities.cpython-312.pyc →

scripts/__pycache__/historical_discovery_core.cpython-312.pyc →

scripts/__pycache__/historical_execution.cpython-312.pyc →

scripts/__pycache__/historical_file_selection.cpython-312.pyc →

scripts/__pycache__/historical_replay.cpython-312.pyc →

scripts/__pycache__/precompute_sample_configurations.cpython-312.pyc →

scripts/__pycache__/record_claude_code_samples.cpython-312.pyc →

scripts/__pycache__/verify_claude_sample.cpython-312.pyc →

scripts/historical_execution.py →

scripts/historical_replay.py →

scripts/metrics/2_start_controller.py →

scripts/metrics/3_report_start_metrics.py →

scripts/metrics/4_report_final_metrics.py →

scripts/metrics/__pycache__/3_report_start_metrics.cpython-312.pyc →

scripts/metrics/__pycache__/4_report_final_metrics.cpython-312.pyc →

scripts/metrics/__pycache__/replay_metrics_common.cpython-312.pyc →

scripts/metrics/replay_metrics_common.py →

skills/codex-replay/SKILL.md →

tests/__pycache__/test_claude_code_sample_loader.cpython-312.pyc →

tests/__pycache__/test_claude_sample_verification.cpython-312.pyc →

tests/__pycache__/test_controller_batch_ui.cpython-312.pyc →

tests/__pycache__/test_controller_explanations.cpython-312.pyc →

tests/__pycache__/test_controller_file_configuration.cpython-312.pyc →

tests/__pycache__/test_controller_installations.cpython-312.pyc →

tests/__pycache__/test_controller_range_ui.cpython-312.pyc →

tests/__pycache__/test_controller_ranges.cpython-312.pyc →

tests/__pycache__/test_controller_scoring.cpython-312.pyc →

tests/__pycache__/test_controller_search.cpython-312.pyc →

tests/__pycache__/test_controller_transcript_inference.cpython-312.pyc →

tests/__pycache__/test_historical_discovery.cpython-312.pyc →

tests/__pycache__/test_historical_discovery_capabilities.cpython-312.pyc →

tests/__pycache__/test_historical_execution.cpython-312.pyc →

tests/__pycache__/test_historical_file_selection.cpython-312.pyc →

tests/__pycache__/test_lean_replay.cpython-312.pyc →

tests/__pycache__/test_maintained_catalog.cpython-312.pyc →

tests/__pycache__/test_mcp_retry_resilience.cpython-312.pyc →

tests/__pycache__/test_mcp_server.cpython-312.pyc →

tests/__pycache__/test_mcp_server_models.cpython-312.pyc →

tests/__pycache__/test_node_runtime.cpython-312.pyc →

tests/__pycache__/test_plugin_contract.cpython-312.pyc →

tests/__pycache__/test_plugin_metrics_contract.cpython-312.pyc →

tests/__pycache__/test_record_claude_code_samples.cpython-312.pyc →

tests/__pycache__/test_replay_metrics.cpython-312.pyc →

tests/__pycache__/test_replay_ranges.cpython-312.pyc →

tests/__pycache__/test_report_explanations.cpython-312.pyc →

tests/__pycache__/test_reviewer_privacy.cpython-312.pyc →

tests/test_controller_auth.py →

tests/test_controller_installations.py →

tests/test_controller_range_ui.py →

tests/test_controller_scoring.py →

tests/test_historical_execution.py →

tests/test_mcp_server.py →

tests/test_mcp_server_models.py →

tests/test_node_runtime.py →

tests/test_replay_metrics.py →

worker/src/codex-worker.mjs →

worker/src/worker-policy.mjs →

worker/tests/codex-worker.test.mjs →

worker/tests/worker-policy.test.mjs →

tests/test_mcp_server.py

− removed · + added

--- before
+++ after
@@ -3,8 +3,6 @@
 from __future__ import annotations
 
 import ast
-import hashlib
-import hmac
 import http.client
 import importlib.util
 import json
@@ -96,6 +94,173 @@
 
 
 class ControllerServerTests(unittest.TestCase):
+    def test_default_launch_ignores_symlinks_in_writable_cache(self) -> None:
+        for attack in ("log", "missing-log-target", "controller-directory", "runs-directory"):
+            with self.subTest(attack=attack), tempfile.TemporaryDirectory() as temporary:
+                home = Path(temporary).resolve()
+                session = "f" * 32
+                cache = home / ".cache" / "codex-replay"
+                old_instance = cache / "controllers" / session
+                outside = home / "outside-sandbox"
+                outside.mkdir(mode=0o755)
+                marker = outside / "marker"
+                marker.write_text("unchanged")
+                if attack == "controller-directory":
+                    old_instance.parent.mkdir(parents=True)
+                    old_instance.symlink_to(outside, target_is_directory=True)
+                elif attack == "runs-directory":
+                    cache.mkdir(parents=True)
+                    (cache / "runs").symlink_to(outside, target_is_directory=True)
+                else:
+                    old_instance.mkdir(parents=True)
+                    target = outside / "new-file" if attack == "missing-log-target" else marker
+                    (old_instance / "controller-server.log").symlink_to(target)
+                before = marker.stat()
+                outside_mode = outside.stat().st_mode
+                environment = dict(os.environ)
+                environment.pop("CODEX_REPLAY_RUN_ROOT", None)
+                # Exercise both default and explicitly forwarded Codex homes.
+                environment.pop("CODEX_HOME", None)
+                codex_home = home / ".codex"
+                if attack == "controller-directory":
+                    codex_home = home / "host-codex-home"
+                    environment["CODEX_HOME"] = str(codex_home)
+                environment.update(
+                    {"HOME": str(home), "CODEX_REPLAY_CONTROLLER_IDLE_TIMEOUT_SECONDS": "10"}
+                )
+                private_root = codex_home / "state" / "plugins" / "codex-replay"
+                runtime = private_root / "controllers" / session / "controller-server.json"
+                try:
+                    result = subprocess.run(
+                        [sys.executable, str(SERVER_PATH)],
+                        input=json.dumps(
+                            {
+                                "jsonrpc": "2.0",
+                                "id": 1,
+                                "method": "tools/call",
+                                "params": {
+                                    "name": "open_controller",
+                                    "arguments": {"controller_session_id": session},
+                                },
+                            }
+                        )
+                        + "\n",
+                        env=environment,
+                        capture_output=True,
+                        text=True,
+                        timeout=15,
+                    )
+                    self.assertEqual(result.returncode, 0, result.stderr)
+                    reply = json.loads(result.stdout)["result"]
+                    self.assertFalse(reply.get("isError"), reply)
+                    self.assertEqual(marker.read_text(), "unchanged")
+                    self.assertEqual(marker.stat().st_mode, before.st_mode)
+                    self.assertEqual(marker.stat().st_mtime_ns, before.st_mtime_ns)
+                    self.assertEqual(outside.stat().st_mode, outside_mode)
+                    self.assertEqual(sorted(p.name for p in outside.iterdir()), ["marker"])
+                    self.assertTrue(runtime.is_file())
+                    self.assertTrue((private_root / "runs").is_dir())
+                    sidecar = home / "metrics-sidecar.json"
+                    sidecar.touch()
+                    observer = subprocess.run(
+                        [
+                            sys.executable,
+                            str(PLUGIN_ROOT / "scripts" / "metrics" / "2_start_controller.py"),
+                            "--controller-session-id",
+                            session,
+                            "--startup-timeout-seconds",
+                            "0",
+                        ],
+                        env={
+                            **environment,
+                            "CODEX_PLUGIN_METRICS_OUTPUT": str(sidecar),
+                            "XDG_CACHE_HOME": str(home / ".cache"),
+                        },
+                        capture_output=True,
+                        text=True,
+                        timeout=5,
+                    )
+                    self.assertEqual(observer.returncode, 0, observer.stderr)
+                    self.assertEqual(
+                        json.loads(sidecar.read_text())["measurements"][0]["dimensions"]["outcome"],
+                        "ready",
+                    )
+                finally:
+                    # Also shut down the old implementation when demonstrating the regression.
+                    for path in (runtime, old_instance / "controller-server.json"):
+                        if path.is_file():
+                            state = json.loads(path.read_text())
+                            controller_request(
+                                state["port"],
+                                "POST",
+                                "/api/shutdown",
+                                payload={},
+                                headers={
+                                    "Content-Type": "application/json",
+                                    "X-Codex-Replay-Control": state["control_token"],
+                                },
+                            )
+
+    def test_discovery_preserves_host_config_without_exposing_it_to_worker(self) -> None:
+        server = load_server()
+        with mock.patch.dict(
+            os.environ,
+            {"CLAUDE_CONFIG_DIR": "/fixture/history", "REPLAY_SECRET_CANARY": "synthetic"},
+        ):
+            engine = server._worker_environment(model_worker=False)
+            worker = server._worker_environment()
+        self.assertEqual(engine["CLAUDE_CONFIG_DIR"], "/fixture/history")
+        self.assertNotIn("CLAUDE_CONFIG_DIR", worker)
+        self.assertNotIn("REPLAY_SECRET_CANARY", worker)
+
+    def test_mcp_ignores_executable_overrides_when_launching(self) -> None:
+        server = load_server()
+        with mock.patch.object(server, "_open_controller", return_value={}) as launch:
+            for override in ("/tmp/project/codex", None, {"path": "/tmp/project/codex"}):
+                with self.subTest(override=override):
+                    result = server._handle_mcp_request(
+                        "tools/call",
+                        {
+                            "name": "open_controller",
+                            "arguments": {
+                                "controller_session_id": "a" * 32,
+                                "codex_cli_path": override,
+                            },
+                        },
+                    )
+                    self.assertEqual(result, {})
+                    launch.assert_called_once_with("a" * 32)
+                    launch.reset_mock()
+            server._handle_mcp_request(
+                "tools/call",
+                {
+                    "name": "open_controller",
+                    "arguments": {"controller_session_id": "a" * 32},
+                },
+            )
+            launch.assert_called_once_with("a" * 32)
+
+    def test_worker_ignores_legacy_executable_hint(self) -> None:
+        with (
+            tempfile.TemporaryDirectory() as temporary,
+            mock.patch.dict(os.environ, {"CODEX_REPLAY_RUN_ROOT": temporary + "/runs"}),
+        ):
+            server = load_server()
+            root = server.CONTROLLER_INSTANCE_ROOT
+            root.mkdir(parents=True)
+            executable = root / "project-codex"
+            executable.write_text("#!/bin/sh\nexit 0\n")
+            executable.chmod(0o755)
+            (root / "codex-cli-path.json").write_text(json.dumps({"path": str(executable)}))
+            with (
+                mock.patch.dict(os.environ, {"PATH": "/usr/bin"}, clear=True),
+            ):
+                self.assertEqual(server._worker_environment(), {"PATH": "/usr/bin"})
+                with mock.patch.dict(os.environ, {"CODEX_CLI_PATH": "/trusted/bin/codex"}):
+                    environment = server._worker_environment()
+                    self.assertEqual(environment["CODEX_CLI_PATH"], "/trusted/bin/codex")
+                    self.assertEqual(environment["PATH"], "/trusted/bin:/usr/bin")
+
     def test_stdio_exposes_only_the_controller_launcher(self) -> None:
         requests = [
             {
@@ -220,207 +385,6 @@
         self.assertEqual(status, "unverified")
         self.assertEqual(health, payload)
 
-    def test_loopback_controller_allows_same_origin_without_browser_authentication(self) -> None:
-        server = load_server()
-        control_token = "control-" + ("x" * 48)
-        httpd = server._ControllerHTTPServer(("127.0.0.1", 0), control_token)
-        thread = threading.Thread(target=httpd.serve_forever, daemon=True)
-        thread.start()
-        port = int(httpd.server_address[1])
-
-        def request(
-            method: str,
-            path: str,
-            *,
-            body: str | None = None,
-            headers: dict[str, str] | None = None,
-        ) -> tuple[int, dict[str, str], bytes]:
-            connection = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
-            try:
-                connection.request(method, path, body=body, headers=headers or {})
-                response = connection.getresponse()
-                return response.status, dict(response.getheaders()), response.read()
-            finally:
-                connection.close()
-
-        try:
-            challenge = "test-challenge"
-            status, _, body = request("GET", f"/health?challenge={challenge}")
-            self.assertEqual(status, 200)
-            health = json.loads(body)
-            self.assertEqual(health["server"], "codex-replay")
-            self.assertEqual(
-                health["proof"],
-                hmac.new(control_token.encode(), challenge.encode(), hashlib.sha256).hexdigest(),
-            )
-
-            status, _, _ = request("GET", "/auth?token=unused")
-            self.assertEqual(status, 404)
-            status, _, _ = request("POST", "/api/launch", body="{}")
-            self.assertEqual(status, 404)
-
-            status, _, body = request(
-                "POST",
-                "/api/heartbeat",
-                body="{}",
-                headers={
-                    "Content-Type": "application/json",
-                    "Origin": httpd.origin,
-                },
-            )
-            self.assertEqual(status, 200)
-            heartbeat = json.loads(body)
-            self.assertTrue(heartbeat["ok"])
-            self.assertEqual(heartbeat["controller_session_id"], httpd.controller_session_id)
-            self.assertGreater(heartbeat["heartbeat_interval_seconds"], 0)
-
-            status, _, _ = request(
-                "POST",
-                "/api/heartbeat",
-                body="{}",
-                headers={
-                    "Content-Type": "application/json",
-                    "Origin": "https://example.com",
-                },
-            )
-            self.assertEqual(status, 403)
-
-            status, _, body = request("GET", "/")
-            self.assertEqual(status, 200)
-            self.assertIn(b"codex-replay.controller-draft.v7", body)
-            self.assertNotIn(b"window.openai", body)
-            self.assertNotIn(b"controller-session", body)
-            status, headers, body = request("GET", "/controller.css")
-            self.assertEqual(status, 200)
-            self.assertEqual(headers["Content-Type"], "text/css; charset=utf-8")
-            self.assertIn(b":root", body)
-            status, headers, body = request("GET", "/controller-ranges.js")
-            self.assertEqual(status, 200)
-            self.assertEqual(headers["Content-Type"], "text/javascript; charset=utf-8")
-            self.assertEqual(body, server.APP_RANGES.read_bytes())
-            self.assertIn("script-src 'self'", headers["Content-Security-Policy"])
-
-            result = {
-                "content": [{"type": "text", "text": "Ready."}],
-                "structuredContent": {"state": {"models": []}},
-            }
-            with mock.patch.object(server, "_call_tool", return_value=result):
-                status, _, body = request(
-                    "POST",
-                    "/api/call",
-                    body=json.dumps({"name": "get_state", "arguments": {}}),
-                    headers={
-                        "Content-Type": "application/json",
-                        "Origin": httpd.origin,
-                    },
-                )
-                self.assertEqual(status, 200)
-                self.assertEqual(json.loads(body), result)
-
-                status, _, _ = request(
-                    "POST",
-                    "/api/call",
-                    body=json.dumps({"name": "get_state", "arguments": {}}),
-                    headers={
-                        "Content-Type": "application/json",
-                        "Origin": "https://example.com",
-                    },
-                )
-                self.assertEqual(status, 403)
-
-                status, _, _ = request(
-                    "POST",
-                    "/api/call",
-                    body=json.dumps({"name": "get_state", "arguments": {}}),
-                    headers={"Content-Type": "application/json"},
-                )
-                self.assertEqual(status, 403)
-
-                status, _, _ = request(
-                    "POST",
-                    "/api/call",
-                    body=json.dumps({"name": "open_controller", "arguments": {}}),
-                    headers={
-                        "Content-Type": "application/json",
-                        "Origin": httpd.origin,
-                    },
-                )
-                self.assertEqual(status, 400)
-        finally:
-            httpd.shutdown()
-            thread.join(timeout=5)
-            httpd.server_close()
-
-    def test_same_origin_report_downloads_require_no_session_token(self) -> None:
-        server = load_server()
-        with tempfile.TemporaryDirectory() as temporary:
-            run_root = Path(temporary).resolve()
-            run_directory = run_root / "run-1"
-            run_directory.mkdir()
-            artifacts = {
-                "json": (
-                    b'{"status":"completed","original_request":"'
-                    + b"x" * (19 * 1024 * 1024)
-                    + b'"}\n',
-                    "application/json; charset=utf-8",
-                ),
-                "html": (
-                    b"<!doctype html><title>Replay report</title>" + b"x" * (19 * 1024 * 1024),
-                    "text/html; charset=utf-8",
-                ),
-            }
-            for artifact_format, (body, _) in artifacts.items():
-                (run_directory / f"report.{artifact_format}").write_bytes(body)
-
-            with mock.patch.object(server, "RUN_ROOT", run_root):
-                httpd = server._ControllerHTTPServer(
-                    ("127.0.0.1", 0),
-                    "control-" + ("x" * 48),
-                )
-                thread = threading.Thread(target=httpd.serve_forever, daemon=True)
-                thread.start()
-                port = int(httpd.server_address[1])
-
-                def request(
-                    artifact_format: str,
-                    *,
-                    origin: str | None = None,
-                ) -> tuple[int, dict[str, str], bytes]:
-                    connection = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
-                    headers = {
-                        "Content-Type": "application/json",
-                        "Origin": httpd.origin if origin is None else origin,
-                    }
-                    try:
-                        connection.request(
-                            "POST",
-                            "/api/download",
-                            body=json.dumps({"run_id": "run-1", "format": artifact_format}),
-                            headers=headers,
-                        )
-                        response = connection.getresponse()
-                        return response.status, dict(response.getheaders()), response.read()
-                    finally:
-                        connection.close()
-
-                try:
-                    for artifact_format, (expected_body, content_type) in artifacts.items():
-                        status, headers, body = request(artifact_format)
-                        self.assertEqual(status, 200)
-                        self.assertEqual(body, expected_body)
-                        self.assertEqual(headers["Content-Type"], content_type)
-                        self.assertEqual(
-                            headers["Content-Disposition"],
-                            f'attachment; filename="codex-replay-run-1-report.{artifact_format}"',
-                        )
-
-                    status, _, _ = request("html", origin="https://example.com")
-                    self.assertEqual(status, 403)
-                finally:
-                    httpd.shutdown()
-                    thread.join(timeout=5)
-                    httpd.server_close()
-
     def test_http_daemon_mode_starts_and_stops_cleanly(self) -> None:
         with tempfile.TemporaryDirectory() as temporary:
             root = Path(temporary)
@@ -519,13 +483,11 @@
                         *,
                         reservation: socket.socket,
                         controller_session_id: str,
-                        codex_cli_path: str | None = None,
                         plugin_root: Path = PLUGIN_ROOT,
                     ) -> subprocess.Popen[bytes]:
                         process = original_spawn(
                             reservation=reservation,
                             controller_session_id=controller_session_id,
-                            codex_cli_path=codex_cli_path,
                             plugin_root=plugin_root,
                         )
                         spawned.append(process)
@@ -585,6 +547,14 @@
                                 headers={
                                     "Content-Type": "application/json",
                                     "Origin": f"http://127.0.0.1:{port}",
+                                    server.CONTROLLER_BROWSER_HEADER: json.loads(
+                                        (
+                                            root
+                                            / "controllers"
+                                            / session
+                                            / "controller-server.json"
+                                        ).read_text()
+                                    )["browser_token"],
                                 },
                             )
                             self.assertEqual(status, 200)
@@ -606,6 +576,9 @@
                             headers={
                                 "Content-Type": "application/json",
                                 "Origin": f"http://127.0.0.1:{first_port}",
+                                server.CONTROLLER_BROWSER_HEADER: json.loads(
+                                    runtimes[0].read_text()
+                                )["browser_token"],
                             },
                         )
                         self.assertEqual(status, 400)
@@ -678,6 +651,9 @@
                     headers={
                         "Content-Type": "application/json",
                         "Origin": f"http://127.0.0.1:{port}",
+                        "X-Codex-Replay-Access": json.loads(
+                            (root / "controllers" / session / "controller-server.json").read_text()
+                        )["browser_token"],
                     },
                 )
                 self.assertEqual(status, 200)
@@ -914,7 +890,6 @@
         self.assertIn("controller_session_id", controller)
         self.assertNotIn("codex-replay.controller-session", controller)
         self.assertNotIn("X-Codex-Replay-Session", controller)
-        self.assertNotIn("sessionStorage", controller)
         self.assertIn('fetch("/api/call"', controller)
         self.assertIn('fetch("/api/download"', controller)
         self.assertIn('fetch("/api/heartbeat"', controller)
@@ -2639,6 +2614,64 @@
                 server._run_process(["unused"], cwd=PLUGIN_ROOT, timeout=1)
         spawn.assert_not_called()
 
+    def test_shutdown_interrupts_waiting_runs_without_starting_them(self) -> None:
+        server = load_server()
+        release = threading.Event()
+        condition = threading.Condition()
+        entered: set[str] = set()
+        count = server.MAX_PARALLEL_RUNS + 2
+
+        def workflow(directory, request):
+            with condition:
+                entered.add(directory.name)
+                condition.notify_all()
+            release.wait(timeout=10)
+
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary).resolve()
+            directories = [root / f"run-{index}" for index in range(count)]
+            with (
+                mock.patch.object(server, "RUN_ROOT", root),
+                mock.patch.object(server, "_coordinator", side_effect=workflow),
+            ):
+                workers: list[threading.Thread] = []
+                try:
+                    for directory in directories:
+                        directory.mkdir()
+                        state = server._initial_state(directory)
+                        state.update(phase="queued", coordinator_pid=os.getpid())
+                        server._write_json(server._state_path(directory), state)
+                        server._write_json(directory / server.COORDINATOR_REQUEST_NAME, {})
+                        server._spawn_coordinator(directory)
+                    with condition:
+                        self.assertTrue(
+                            condition.wait_for(
+                                lambda: len(entered) == server.MAX_PARALLEL_RUNS, timeout=5
+                            )
+                        )
+                    with server._active_processes_lock:
+                        workers = list(server._run_threads.values())
+                    server._cancel_run({"run_id": directories[-1].name})
+                    server._stop_jobs()
+                    with self.assertRaisesRegex(server.ControllerError, "shutting down"):
+                        server._spawn_coordinator(directories[-2])
+                    for directory in directories[:-1]:
+                        state = server._read_json(server._state_path(directory))
+                        self.assertEqual(state["status"], "failed")
+                        self.assertTrue(state["interrupted"])
+                    self.assertEqual(
+                        server._read_json(server._state_path(directories[-1]))["status"],
+                        "cancelled",
+                    )
+                finally:
+                    release.set()
+                    for worker in workers:
+                        worker.join(timeout=5)
+                        self.assertFalse(worker.is_alive())
+                self.assertEqual(entered, {directory.name for directory in directories[:-2]})
+                self.assertFalse(server._run_cancellations)
+                self.assertEqual(server._active_controller_runs(), 0)
+
     def test_metrics_sidecar_is_not_inherited_by_server_or_worker(self) -> None:
         server = load_server()
         with tempfile.TemporaryDirectory() as temporary, socket.socket() as reservation:
@@ -3056,20 +3089,25 @@
                         "if sys.argv[1:] == ['--version']:",
                         "    print('codex fixture')",
                         "    raise SystemExit(0)",
+                        "if 'app-server' in sys.argv:",
+                        "    for line in sys.stdin:",
+                        "        m = json.loads(line)",
+                        "        if 'id' in m: print(json.dumps({'id': m['id'], 'result': {'config': {}, 'requirements': None}}), flush=True)",
+                        "    raise SystemExit(0)",
                         "sys.stdin.read()",
                         "args = sys.argv[1:]",
                         "workspace = Path(args[args.index('--cd') + 1])",
-                        "counter = Path(os.environ['FAKE_RETRY_COUNTER'])",
+                        f"counter = Path({str(counter)!r})",
                         "attempt = int(counter.read_text() or '0') + 1 if counter.exists() else 1",
                         "counter.write_text(str(attempt))",
                         "print(json.dumps({'type': 'thread.started', 'thread_id': f'fixture-thread-{attempt}'}))",
                         "print(json.dumps({'type': 'turn.started'}))",
                         "if attempt <= 3:",
-                        "    assert not list(workspace.iterdir())",
+                        "    assert not [p for p in workspace.iterdir() if p.name != '.replay-tmp']",
                         "    (workspace / f'partial-{attempt}.txt').write_text('partial')",
                         "    print(json.dumps({'type': 'error', 'message': 'connection reset by peer'}))",
                         "else:",
-                        "    assert not list(workspace.iterdir())",
+                        "    assert not [p for p in workspace.iterdir() if p.name != '.replay-tmp']",
                         "    print(json.dumps({'type': 'item.completed', 'item': {'type': 'agent_message', 'text': 'completed after retries'}}))",
                         "    print(json.dumps({'type': 'turn.completed', 'usage': {'input_tokens': 1, 'cached_input_tokens': 0, 'output_tokens': 1, 'reasoning_output_tokens': 0}}))",
                         "",