From 72f9d8b398aa0cb00a8fcf03e34bd7e0ad08610c Mon Sep 17 00:00:00 2001 From: psv666 <2693925048@qq.com> Date: Tue, 27 Oct 2026 11:23:45 +0800 Subject: [PATCH] fix(benchmarks): preserve interrupted trace replay evidence Signed-off-by: psv666 <2693925048@qq.com> --- benchmarks/traces/README.md | 31 ++- benchmarks/traces/replay.py | 157 ++++++++---- benchmarks/traces/results-2026-10-25.md | 63 ----- .../test_trace_replay_persistence.py | 239 ++++++++++++++++++ 4 files changed, 373 insertions(+), 117 deletions(-) delete mode 100644 benchmarks/traces/results-2026-10-25.md create mode 100644 tests/benchmarks/test_trace_replay_persistence.py diff --git a/benchmarks/traces/README.md b/benchmarks/traces/README.md index 35fc9387..3fc57593 100644 --- a/benchmarks/traces/README.md +++ b/benchmarks/traces/README.md @@ -10,7 +10,10 @@ it repeats, and replay it at the pace the agent sent it. - [`replay.py`](replay.py) replays a trace against a running server, one client per episode, keeping its order and think time. - [`samples/`](samples/) holds two small traces whose content may be shared. -- [The first results](results-2026-10-25.md) come from traces recorded on 2026-10-25. +- The [historical report from 2026-10-25](https://github.com/InferMatrixLab/system1-omni/blob/c18e9e7c634412e623cfa0f18aed25809c3d75a0/benchmarks/traces/results-2026-10-25.md) + remains available at the original revision for [PR #185](https://github.com/InferMatrixLab/system1-omni/pull/185). + It is author-reported experiment evidence, not a maintained benchmark input or a + validation of the current revision. They need Python 3.11+ and `httpx==0.28.1`; `metrics.py` also needs `tokenizers`. @@ -90,8 +93,30 @@ misses count answered requests only. The output has `config.json` (with the trac checksum), `responses.jsonl` (latency, status, deadline miss, `prefix_cache`, base64 body) and `summary.json` (median and nearest-rank p95, as `bench.py` reports them). +`--out` must name a new directory; an existing directory, even an empty one, is +rejected before sending requests or modifying its contents. `config.json` lists +all planned request IDs after `--skip-retries` filtering. Each finished attempt is +appended and flushed to `responses.jsonl` immediately, in completion order; use +`sent_s` to recover send order. On Ctrl-C or a replay error, all episode tasks stop +before the files are finalized, and the process exits unsuccessfully. Every +planned request then has a row with one of these states: + +- `completed`: an HTTP response or transport error was observed; +- `incomplete`: it was attempted but interrupted before its response completed; +- `unattempted`: it was still waiting for its episode start or next step. + +Incomplete and unattempted rows have no response latency or deadline result. +The summary separates `completed`, `incomplete`, and `unattempted`; `failed` +counts completed HTTP/transport failures, and `ok` counts completed HTTP 200 +responses. It also records `stop_reason` (`completed`, `interrupted`, or `error`). +Latency/deadline statistics exclude unfinished work, while `requests` retains the +full planned denominator. A forced kill or machine failure cannot run finalization: +keep the already-flushed responses and config as partial evidence, not a completed +run. A flush is not a power-loss durability guarantee. + ## Data Traces of real web pages copy third-party content. Keep them outside this repository -and commit only derived counts, or traces from local fixtures, self-hosted pages or -sets whose license allows it. +and retain raw run output and derived reports in PR/CI evidence or durable artifact +archives. The source tree should contain maintained replay inputs or fixtures from +local fixtures, self-hosted pages or sets whose license allows it. diff --git a/benchmarks/traces/replay.py b/benchmarks/traces/replay.py index 33fa41cf..e428a2fe 100644 --- a/benchmarks/traces/replay.py +++ b/benchmarks/traces/replay.py @@ -47,7 +47,7 @@ def offsets(groups, together): return {g: (v - first) / 1000 for g, v in starts.items()} -async def run_episode(client, url, model, records, delay, scale, clock, sink, deadline_s=None): +async def run_episode(client, url, model, records, delay, scale, clock, results, save): await asyncio.sleep(delay) for i, record in enumerate(records): if i: @@ -56,39 +56,50 @@ async def run_episode(client, url, model, records, delay, scale, clock, sink, de if model: body["model"] = model sent = time.perf_counter() - error = None + result = results[record["id"]] + result.update(state="incomplete", sent_s=round(sent - clock, 6)) try: - response = await client.post(url, json=body) - status, content = response.status_code, response.content - except httpx.HTTPError as e: - status, content, error = None, b"", f"{type(e).__name__}: {e}" - latency = (time.perf_counter() - sent) * 1000 - prefix_cache = None - if status == 200: + error = None try: - prefix_cache = json.loads(content).get("metadata", {}).get("prefix_cache") - except (ValueError, AttributeError): - pass - deadline = deadline_s if deadline_s is not None else record.get("deadline_s") - answered = status == 200 - sink.append({ - "id": record["id"], "group": record["group"], "step": record["step"], - "attempt": record.get("attempt", 0), "sent_s": round(sent - clock, 6), - "latency_ms": round(latency, 3), "status": status, "error": error, - "deadline_s": deadline, - "missed": latency > deadline * 1000 if answered and deadline is not None else None, - "prefix_cache": prefix_cache, "body": base64.b64encode(content).decode(), - }) + response = await client.post(url, json=body) + status, content = response.status_code, response.content + except httpx.HTTPError as e: + status, content, error = None, b"", f"{type(e).__name__}: {e}" + latency = (time.perf_counter() - sent) * 1000 + prefix_cache = None + if status == 200: + try: + prefix_cache = json.loads(content).get("metadata", {}).get("prefix_cache") + except (ValueError, AttributeError): + pass + deadline = result["deadline_s"] + result.update( + state="completed", latency_ms=round(latency, 3), status=status, error=error, + missed=latency > deadline * 1000 if status == 200 and deadline is not None else None, + prefix_cache=prefix_cache, body=base64.b64encode(content).decode(), + ) + except asyncio.CancelledError: + result["error"] = "request cancelled before its response completed" + raise + except Exception as exc: + result["error"] = f"{type(exc).__name__}: {exc}" + raise + finally: + save(result) def summarize(records, elapsed): """Latency of answered requests, with `bench.py`'s median and nearest-rank p95.""" - ok = [r for r in records if r["status"] == 200] + completed = [r for r in records if r["state"] == "completed"] + ok = [r for r in completed if r["status"] == 200] latencies = sorted(r["latency_ms"] for r in ok) with_deadline = [r for r in ok if r["missed"] is not None] - shared = [r for r in ok if (r["prefix_cache"] or {}).get("enabled")] + shared = [r for r in ok if isinstance(r["prefix_cache"], dict) and r["prefix_cache"].get("enabled")] return { - "requests": len(records), "ok": len(ok), "failed": len(records) - len(ok), + "requests": len(records), "ok": len(ok), "failed": len(completed) - len(ok), + "completed": len(completed), + "incomplete": sum(r["state"] == "incomplete" for r in records), + "unattempted": sum(r["state"] == "unattempted" for r in records), "elapsed_s": round(elapsed, 3), "latency_ms": { "p50": statistics.median(latencies) if latencies else None, @@ -107,19 +118,70 @@ async def replay(args): groups = episodes(requests, args.skip_retries) starts = offsets(groups, args.together) url = args.url.rstrip("/") + "/v1/systemone" - sink = [] + results = { + r["id"]: { + "id": r["id"], "group": r["group"], "step": r["step"], "attempt": r.get("attempt", 0), + "state": "unattempted", "sent_s": None, "latency_ms": None, "status": None, + "error": None, "deadline_s": args.deadline_s if args.deadline_s is not None else r.get("deadline_s"), + "missed": None, "prefix_cache": None, "body": "", + } + for group in groups.values() for r in group + } + config = { + "trace": str(args.trace), "trace_sha256": digest(args.trace), "url": args.url, + "model": args.model, "think_scale": args.think_scale, "together": args.together, + "sequential": args.sequential, "deadline_s": args.deadline_s, + "skip_retries": args.skip_retries, "timeout_s": args.timeout, + "response_format": RESPONSE_FORMAT, "python": platform.python_version(), + "httpx": httpx.__version__, "planned_ids": list(results), + } + # mkdir claims the whole run atomically, including against another replay process. + # Even an empty existing directory may belong to a run that has just started. + args.out.mkdir(parents=True) + write_json(args.out / "config.json", config) + saved = set() clock = time.perf_counter() - limits = httpx.Limits(max_connections=len(groups), max_keepalive_connections=len(groups)) - async with httpx.AsyncClient(timeout=args.timeout, trust_env=False, limits=limits) as client: - if args.sequential: - for records in groups.values(): - await run_episode(client, url, args.model, records, 0.0, args.think_scale, clock, sink, args.deadline_s) - else: - await asyncio.gather(*( - run_episode(client, url, args.model, records, starts[g], args.think_scale, clock, sink, args.deadline_s) - for g, records in groups.items() - )) - return sink, time.perf_counter() - clock + stop_reason = "error" + with (args.out / "responses.jsonl").open("x", encoding="utf-8") as responses: + def save(result): + responses.write(json.dumps(result, allow_nan=False) + "\n") + responses.flush() + saved.add(result["id"]) + + limits = httpx.Limits(max_connections=len(groups), max_keepalive_connections=len(groups)) + try: + async with httpx.AsyncClient(timeout=args.timeout, trust_env=False, limits=limits) as client: + tasks = [] + try: + if args.sequential: + for records in groups.values(): + await run_episode(client, url, args.model, records, 0.0, args.think_scale, + clock, results, save) + else: + tasks = [asyncio.create_task(run_episode( + client, url, args.model, records, starts[g], args.think_scale, + clock, results, save, + )) for g, records in groups.items()] + await asyncio.gather(*tasks) + finally: + # gather propagates one failure without stopping its siblings. Drain + # every episode before closing the client or finalizing the evidence. + for task in tasks: + if not task.done(): + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + stop_reason = "completed" + except asyncio.CancelledError: + stop_reason = "interrupted" + raise + finally: + for result in results.values(): + if result["id"] not in saved: + save(result) + summary = summarize(list(results.values()), time.perf_counter() - clock) + summary["stop_reason"] = stop_reason + write_json(args.out / "summary.json", summary) + return summary def main(): @@ -139,20 +201,13 @@ def main(): parser.error("--think-scale must be nonnegative") if args.sequential and args.together: parser.error("--sequential and --together exclude each other") - args.out.mkdir(parents=True, exist_ok=True) - write_json(args.out / "config.json", { - "trace": str(args.trace), "trace_sha256": digest(args.trace), "url": args.url, - "model": args.model, "think_scale": args.think_scale, "together": args.together, - "sequential": args.sequential, "deadline_s": args.deadline_s, - "skip_retries": args.skip_retries, "timeout_s": args.timeout, - "response_format": RESPONSE_FORMAT, "python": platform.python_version(), - "httpx": httpx.__version__, - }) - records, elapsed = asyncio.run(replay(args)) - records.sort(key=lambda r: r["sent_s"]) - (args.out / "responses.jsonl").write_text("".join(json.dumps(r) + "\n" for r in records)) - summary = summarize(records, elapsed) - write_json(args.out / "summary.json", summary) + try: + summary = asyncio.run(replay(args)) + except FileExistsError: + parser.error(f"output directory already exists: {args.out}; choose a new --out directory") + except KeyboardInterrupt: + print(f"Replay interrupted; check {args.out} for partial results", file=sys.stderr) + raise SystemExit(130) print(json.dumps(summary, indent=2)) diff --git a/benchmarks/traces/results-2026-10-25.md b/benchmarks/traces/results-2026-10-25.md deleted file mode 100644 index 6b329ea9..00000000 --- a/benchmarks/traces/results-2026-10-25.md +++ /dev/null @@ -1,63 +0,0 @@ -# Repeated work in System1-Agents traces, 2026-10-25 - -Recorded on one NVIDIA RTX 6000 Ada with [`record_s1a.py`](record_s1a.py) from -System1-Agents `9b20b98`. Open-Jev-9B decided: the native worker with #179's change, no -prefix switch, the merged export's default 4,096-token limit. Fronts that need a chat -model used Qwen3-8B through vLLM 0.32 with thinking off. Flights ran against live Google -Flights (`hl=en&gl=CH`, one route per episode); the form fixtures are -`evals.recovery.live`'s three tasks with recovery off and on. Each Flights and form -episode was one invocation with its own `--group`, and the invocations were concatenated -into one trace per front; each keeps its own clock, so replay these with `--sequential`. `zurich-london` is System1-Agents' own recording of a Jev 1.13 -run (`docs/results/laya-browser-shaping/zurich-london.jsonl`), converted with -`tracefile.py from-ticks`. - -Only the [samples](samples/) are in this repository; the other traces copy third-party -pages or prompt text and stay outside it. The ticket-router row can be recomputed from -its sample. - -## Shares of prefill tokens that could be skipped - -| Trace | Episodes / measured decisions (unserved) | Open-Jev: within a request | + server cache, last 8 | Cua-S1: within a request | + server cache, last 8 | -| --- | ---: | ---: | ---: | ---: | ---: | -| Google Flights, live | 28 / 93 (5) | 89.2% | 90.8% | 4.4% | 31.8% | -| `zurich-london` | 1 / 10 (2); Cua-S1 9 (3) | 91.5% | 91.9% | 4.7% | 35.6% | -| Form fixtures | 36 / 108 | 60.5% | 61.6% | 9.5% | 49.4% | -| 2048 | 5 / 800 | 58.2% | 59.7% | 0% | 46.6% | -| Ticket router | 1 / 30 | 59.8% | 59.8% | 0% | 45.3% | -| Blackjack | 100 / 119 | 41.8% | 41.8% | 0% | 77.2% | - -Each layout is measured on its own tokens. Unserved requests are those recorded as -failed (the five Flights results pages over 4,096 tokens) or that the worker would refuse -(two `zurich-london` pages over Open-Jev-9B's limit; three with more than 26 click -targets for Cua-S1). The Cua-S1 "within a request" share is what Decider's planner would -skip; the Cua-S1 worker runs none. - -- No state came back exactly in any trace. -- Where the state comes first (Open-Jev), sharing within a request is nearly all of the - reuse; a cache of recent prompts adds at most 1.6 points. -- Where each question's goal comes first (Cua-S1), one sequence per question leaves - little to share within a request, but the system prompt and the goal repeat at every - step, so a cache of recent prompts could skip 32 to 77%. Blackjack's 77% needs a - server-side cache that ignores episode boundaries: each hand is its own short episode - (84 of 100 take one decision); within episodes it is 12%. - -## Pace - -| Trace | Think time p50 / p90 | Decision latency p50 / p90 (recorded) | -| --- | ---: | ---: | -| Google Flights | 0.67 / 6.7 s | 6.4 / 7.7 s | -| Form fixtures | 1.2 / 3.4 s | 1.07 / 1.27 s | -| 2048 | 0.67 / 0.68 s | 0.14 / 0.15 s | -| Ticket router, blackjack (no chat model) | 1 to 3 ms | 0.08 to 0.23 s | - -Agents send one request at a time; without a chat model the next request follows the -last response within milliseconds. - -## Caveats - -- Trajectories depend on the deciding model. Open-Jev-9B ends most Flights episodes at - the second decision (22 of 28), as System1-Agents' A40 runs also show, so Flights - episodes are short. -- Flights results pages can exceed the 9B export's 4,096-token default; the export can be - made with a longer `--max-length`. -- One GPU, one decision model, one chat model, one day. diff --git a/tests/benchmarks/test_trace_replay_persistence.py b/tests/benchmarks/test_trace_replay_persistence.py new file mode 100644 index 00000000..fe86d31d --- /dev/null +++ b/tests/benchmarks/test_trace_replay_persistence.py @@ -0,0 +1,239 @@ +"""CPU-only replay persistence regressions using a local HTTP server and mock transport.""" + +import argparse +import asyncio +import base64 +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +from pathlib import Path +import signal +import subprocess +import sys +import tempfile +import threading +import unittest +from unittest.mock import patch + +import httpx + +TRACES = Path(__file__).resolve().parents[2] / "benchmarks/traces" +sys.path.insert(0, str(TRACES)) +import replay # noqa: E402 +import tracefile # noqa: E402 + + +def request(group, step, state, **extra): + return {"type": "request", "id": f"{group}:{step}", "group": group, "step": step, + "request": {"state": state, "questions": {}}, **extra} + + +def write_trace(path, requests): + tracefile.write(path, [{"type": "header", "format": tracefile.FORMAT, + "capture": "converted"}, *requests]) + + +def responses(out): + return [json.loads(line) for line in (out / "responses.jsonl").read_text().splitlines()] + + +class ReplayProcessTests(unittest.TestCase): + def setUp(self): + self.root = Path(self.enterContext(tempfile.TemporaryDirectory())) + self.arrivals = [] + self.held = {name: threading.Event() for name in ("hold-a", "hold-b")} + self.release = threading.Event() + owner = self + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + state = body["state"] + owner.arrivals.append(state) + if state in owner.held: + owner.held[state].set() + owner.release.wait(15) + raw = json.dumps({"echo": state, "metadata": {"prefix_cache": {"enabled": True}}}).encode() + self.send_response(200) + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + try: + self.wfile.write(raw) + except (BrokenPipeError, ConnectionResetError): + pass + + def log_message(self, *unused): + pass + + self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=self.server.serve_forever, daemon=True).start() + self.addCleanup(self.server.server_close) + self.addCleanup(self.server.shutdown) + self.addCleanup(self.release.set) + self.trace, self.out = self.root / "trace.jsonl", self.root / "out" + + def command(self, *flags): + return [sys.executable, str(TRACES / "replay.py"), str(self.trace), "--url", + f"http://127.0.0.1:{self.server.server_port}", "--think-scale", "0", + "--out", str(self.out), *flags] + + def start(self, *flags): + process = subprocess.Popen(self.command(*flags), stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) + + def cleanup(): + if process.poll() is None: + process.kill() + process.communicate(timeout=10) + + self.addCleanup(cleanup) + return process + + def check_interruption(self, sequential): + planned = [request("a", 0, "done"), request("a", 1, "hold-a"), + request("a", 2, "later-a"), request("b", 0, "hold-b"), + request("b", 1, "later-b")] + write_trace(self.trace, planned) + process = self.start("--sequential" if sequential else "--together") + self.assertTrue(self.held["hold-a"].wait(10), "second request never arrived") + if not sequential: + self.assertTrue(self.held["hold-b"].wait(10), "concurrent episode never arrived") + # Capture the live file before cancellation; completion must not depend on exit. + live = responses(self.out) if (self.out / "responses.jsonl").exists() else [] + process.send_signal(signal.SIGINT) + stdout, stderr = process.communicate(timeout=10) + self.assertNotEqual(process.returncode, 0, (stdout, stderr)) + self.assertEqual([row["id"] for row in live], ["a:0"]) + config = json.loads((self.out / "config.json").read_text()) + self.assertCountEqual(config["planned_ids"], [row["id"] for row in planned]) + rows = responses(self.out) + self.assertCountEqual([row["id"] for row in rows], [row["id"] for row in planned]) + by_id = {row["id"]: row for row in rows} + self.assertEqual(by_id["a:0"]["state"], "completed") + self.assertEqual(json.loads(base64.b64decode(by_id["a:0"]["body"]))["echo"], "done") + self.assertEqual(by_id["a:1"]["state"], "incomplete") + self.assertEqual(by_id["b:0"]["state"], "unattempted" if sequential else "incomplete") + for name in ("a:2", "b:1"): + self.assertEqual(by_id[name]["state"], "unattempted") + self.assertIsNone(by_id[name]["sent_s"]) + for row in rows: + if row["state"] != "completed": + for key in ("status", "latency_ms", "missed", "prefix_cache"): + self.assertIsNone(row[key], (row, key)) + summary = json.loads((self.out / "summary.json").read_text()) + expected = {"requests": 5, "ok": 1, "failed": 0, "completed": 1, + "incomplete": 1 if sequential else 2, "unattempted": 3 if sequential else 2, + "stop_reason": "interrupted"} + for key, value in expected.items(): + self.assertEqual(summary[key], value, key) + self.assertEqual(summary["latency_ms"]["mean"], by_id["a:0"]["latency_ms"]) + self.assertEqual(summary["shared_prefix_requests"], 1) + + @unittest.skipUnless(sys.platform != "win32", "requires POSIX SIGINT") + def test_sigint_preserves_sequential_progress(self): + self.check_interruption(sequential=True) + + @unittest.skipUnless(sys.platform != "win32", "requires POSIX SIGINT") + def test_sigint_preserves_all_concurrent_episodes(self): + self.check_interruption(sequential=False) + + def test_existing_directory_is_rejected_without_modifying_any_bytes_or_sending(self): + write_trace(self.trace, [request("a", 0, "done")]) + for name in (None, "config.json", "responses.jsonl", "summary.json"): + with self.subTest(existing=name): + self.out = self.root / (name or "empty") + self.out.mkdir() + if name: + (self.out / name).write_bytes(b"original evidence\n") + before = {p.name: p.read_bytes() for p in self.out.iterdir()} + result = subprocess.run(self.command(), capture_output=True, text=True, timeout=10) + self.assertNotEqual(result.returncode, 0) + self.assertIn("already exists", result.stderr) + self.assertEqual({p.name: p.read_bytes() for p in self.out.iterdir()}, before) + self.assertEqual(self.arrivals, []) + + def test_second_process_cannot_reuse_an_active_run_directory(self): + write_trace(self.trace, [request("a", 0, "hold-a")]) + first = self.start() + self.assertTrue(self.held["hold-a"].wait(10)) + before = {p.name: p.read_bytes() for p in self.out.iterdir()} + second = subprocess.run(self.command(), capture_output=True, text=True, timeout=10) + self.assertNotEqual(second.returncode, 0) + self.assertEqual({p.name: p.read_bytes() for p in self.out.iterdir()}, before) + self.assertEqual(self.arrivals, ["hold-a"]) + self.release.set() + stdout, stderr = first.communicate(timeout=10) + self.assertEqual(first.returncode, 0, (stdout, stderr)) + self.assertEqual(len(responses(self.out)), 1) + + +class ReplayFailureTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.root = Path(self.enterContext(tempfile.TemporaryDirectory())) + self.args = argparse.Namespace(trace=self.root / "trace.jsonl", out=self.root / "out", + url="http://local.test", model=None, think_scale=0, + together=True, sequential=False, skip_retries=False, + deadline_s=None, timeout=1) + + async def test_internal_error_drains_other_episodes_before_saving_summary(self): + started, cancelled = asyncio.Event(), asyncio.Event() + + async def handle(request): + state = json.loads(request.content)["state"] + if state == "crash": + await started.wait() + raise RuntimeError("local transport failed") + started.set() + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + write_trace(self.args.trace, [request("a", 0, "crash"), request("b", 0, "blocked"), + request("b", 1, "later")]) + client = httpx.AsyncClient(transport=httpx.MockTransport(handle)) + with patch.object(replay.httpx, "AsyncClient", return_value=client): + with self.assertRaisesRegex(RuntimeError, "local transport failed"): + await replay.replay(self.args) + self.assertTrue(cancelled.is_set()) + rows = responses(self.args.out) + self.assertEqual(sorted(row["state"] for row in rows), ["incomplete", "incomplete", "unattempted"]) + summary = json.loads((self.args.out / "summary.json").read_text()) + self.assertEqual(summary["stop_reason"], "error") + self.assertEqual((summary["ok"], summary["failed"], summary["incomplete"], summary["unattempted"]), (0, 0, 2, 1)) + + async def test_non_object_cache_metadata_does_not_lose_the_summary(self): + raw = b'{"metadata":{"prefix_cache":true}}' + write_trace(self.args.trace, [request("a", 0, "done")]) + client = httpx.AsyncClient(transport=httpx.MockTransport(lambda req: httpx.Response(200, content=raw))) + with patch.object(replay.httpx, "AsyncClient", return_value=client): + await replay.replay(self.args) + rows = responses(self.args.out) + self.assertEqual(base64.b64decode(rows[0]["body"]), raw) + summary = json.loads((self.args.out / "summary.json").read_text()) + self.assertEqual((summary["ok"], summary["shared_prefix_requests"]), (1, 0)) + + async def test_http_errors_remain_completed_failures_and_skipped_retries_are_not_planned(self): + async def handle(req): + if json.loads(req.content)["state"] == "timeout": + raise httpx.ReadTimeout("timed out", request=req) + return httpx.Response(503, content=b"unavailable") + + retry = dict(request("a", 0, "skip"), id="retry", attempt=1, retry_of="a:0") + write_trace(self.args.trace, [request("a", 0, "http-error"), retry, request("a", 1, "timeout")]) + self.args.skip_retries = True + client = httpx.AsyncClient(transport=httpx.MockTransport(handle)) + with patch.object(replay.httpx, "AsyncClient", return_value=client): + await replay.replay(self.args) + rows = responses(self.args.out) + self.assertEqual([row["id"] for row in rows], ["a:0", "a:1"]) + self.assertTrue(all(row["state"] == "completed" for row in rows)) + self.assertEqual(base64.b64decode(rows[0]["body"]), b"unavailable") + self.assertIn("ReadTimeout", rows[1]["error"]) + summary = json.loads((self.args.out / "summary.json").read_text()) + self.assertEqual((summary["requests"], summary["completed"], summary["failed"], summary["ok"]), (2, 2, 2, 0)) + self.assertEqual(summary["stop_reason"], "completed") + self.assertIsNone(summary["latency_ms"]["mean"]) + + +if __name__ == "__main__": + unittest.main()