From 8a3d03a18b9256dafd814dc1059da49aee8a3a19 Mon Sep 17 00:00:00 2001 From: "Aaron K. Clark (CryptoJones)" Date: Fri, 28 Aug 2026 23:21:46 -0500 Subject: [PATCH] Persist each lane's answer as it lands, not after the last one (#58) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A run interrupted anywhere in the fan-out discarded every answer it had already collected. `list(pool.map(...))` blocks until the last lane returns, and the transcript was written only after that, so all answers lived in memory for the whole run. A SIGKILL, a closed terminal, a harness timeout or a Ctrl-C threw the lot away — including paid http lanes that had already billed for tokens nobody would ever read. It presents as a run that "randomly dies" with zero output and no transcript, and it is worst exactly when it costs most: a long brief with many lanes, where the slowest lane decides whether the rest survive. The transcript path is now claimed before the fan-out and rewritten after every lane completes, driven by as_completed. Each result is written into the slot its lane was dispatched from, so ordering is preserved and a partial transcript reads like a complete one with null for the lanes still out. Writes go to a .tmp and are replaced into place, so a kill mid-write leaves the previous good transcript rather than a truncated one. Partial transcripts carry "partial": true and a note; the final write clears it and attaches the synthesis. Verified: kill -9 mid-run now leaves a transcript holding the two lanes that had answered, third slot null. A completed run is unchanged — no partial marker, synthesis and collisions attached, no stray .tmp. Suite green (119 passed, 29 subtests). Verification set items 1 and 2 re-run by hand. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01VmauK5UCYGRYyXoQ36FQ3S --- BACKLOG.md | 7 ++++++ roundtable | 68 ++++++++++++++++++++++++++++++++++++++++++++---------- 2 files changed, 63 insertions(+), 12 deletions(-) diff --git a/BACKLOG.md b/BACKLOG.md index c93e88b..e9ec40c 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -10,6 +10,13 @@ neither side drifts. poolside streams a full answer — generated, billed, never delivered ([#56](https://github.com/CryptoJones/FlatlineRoundtable/issues/56)) +- [x] A run that dies mid-fan-out lost every answer it already collected + ([#58](https://github.com/CryptoJones/FlatlineRoundtable/issues/58)) — + `list(pool.map(...))` held all answers in memory and wrote the transcript + only after the last lane returned, so any interruption discarded completed + lanes, paid `http` ones included. Now persisted after every lane via + `as_completed`, atomically, marked `"partial": true` until the run finishes. + ## Verification set Run before any PR. Most of these are behavioural and no test suite covers them. diff --git a/roundtable b/roundtable index 6ec1639..3bd99d3 100755 --- a/roundtable +++ b/roundtable @@ -1240,12 +1240,60 @@ def main() -> int: lock, spent = threading.Lock(), [0.0] t0 = time.time() + + # The transcript path is claimed BEFORE the fan-out, not after it, so that a + # run which dies mid-flight still leaves its finished lanes on disk. This + # used to be `list(pool.map(...))` followed by a single write at the end: + # every completed answer was held in memory until the last lane returned, so + # a SIGKILL, a closed terminal, a harness timeout or a Ctrl-C anywhere in the + # fan-out threw away every answer already collected -- including paid http + # lanes that had billed for tokens nobody would ever read. The symptom is a + # run that "randomly dies" with zero output and no transcript. + path = None + if not a.no_transcript: + TRANSCRIPT_DIR.mkdir(parents=True, exist_ok=True) + stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S") + path = TRANSCRIPT_DIR / f"{stamp}.json" + + def save(results_so_far, synthesis=None, collisions=None, partial=True): + """Write the transcript. Called after every lane, and again at the end. + + Atomic via write-then-replace: a kill during the write leaves the previous + good transcript rather than a truncated one. + """ + if path is None: + return + payload = {"brief": prompt, "results": results_so_far, + "synthesis": synthesis, "collisions": collisions} + if partial: + payload["partial"] = True + payload["note"] = ("run did not finish — these are the lanes that had " + "answered when the transcript was last written") + tmp = path.with_suffix(".json.tmp") + tmp.write_text(json.dumps(payload, indent=2)) + tmp.replace(path) + + ordered = [None] * len(lanes) with cf.ThreadPoolExecutor(max_workers=max(len(lanes), 1)) as pool: - results = list(pool.map( - lambda l: ask(l, prompt, keys, int(cfg.get("retries", 1)), deadline, sems, lock, - spent, lane_prices(l, prices), pacers), - lanes, - )) + futures = { + pool.submit(ask, l, prompt, keys, int(cfg.get("retries", 1)), deadline, + sems, lock, spent, lane_prices(l, prices), pacers): i + for i, l in enumerate(lanes) + } + for fut in cf.as_completed(futures): + i = futures[fut] + try: + ordered[i] = fut.result() + except Exception as e: # noqa: BLE001 — mirror ask()'s own contract + ordered[i] = {"lane": lanes[i]["name"], "model": lanes[i].get("model"), + "vendor": lanes[i].get("vendor"), "harness": lanes[i]["harness"], + "answer": None, "error": str(e), "finish_reason": None} + # Persist after every lane. Order is preserved by writing into the + # slot the lane was dispatched from, so a partial transcript reads + # the same way a complete one does, with `null` for lanes still out. + save(ordered) + + results = ordered answered = [r for r in results if r["answer"]] @@ -1267,13 +1315,9 @@ def main() -> int: deadline=deadline, pacers=pacers, retries=int(cfg.get("retries", 1))) - if not a.no_transcript: - TRANSCRIPT_DIR.mkdir(parents=True, exist_ok=True) - stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S") - path = TRANSCRIPT_DIR / f"{stamp}.json" - path.write_text(json.dumps( - {"brief": prompt, "results": results, "synthesis": synth, - "collisions": collisions}, indent=2)) + # Final write: same file the fan-out has been updating, now complete and with + # the synthesis attached. Clears the `partial` marker. + save(results, synthesis=synth, collisions=collisions, partial=False) if a.json: print(json.dumps({"results": results, "synthesis": synth} if a.diff else results, indent=2))