diff --git a/BACKLOG.md b/BACKLOG.md index d3e5e7c..32ddde4 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -19,6 +19,13 @@ neither side drifts. and pre-round hygiene ([#66](https://github.com/CryptoJones/FlatlineRoundtable/issues/66)) +- [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 022c3d4..17b5ef5 100755 --- a/roundtable +++ b/roundtable @@ -1593,13 +1593,71 @@ 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 + # brief_on_record stays the ORIGINAL question in a revision round; in a + # first round it is just `prompt`. A revision transcript also records the + # round number, its parents, and the alias map, exactly as a completed + # round does -- so a partial revision transcript reads like a full one. + payload = {"brief": brief_on_record, "results": results_so_far, + "synthesis": synthesis, "collisions": collisions} + if prompts: + payload.update({"round": round_no, + "parent": [str(p) for p in prior_paths], + "aliases": aliases}) + 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, prompts[l["name"]] if prompts else prompt, - keys, int(cfg.get("retries", 1)), deadline, sems, lock, - spent, lane_prices(l, prices), pacers), - lanes, - )) + # In a revision round each lane gets its own packet (prompts[name]); + # otherwise every lane gets the same brief. + futures = { + pool.submit(ask, l, prompts[l["name"]] if prompts else 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"]] @@ -1625,17 +1683,10 @@ 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" - record = {"brief": brief_on_record, "results": results, "synthesis": synth, - "collisions": collisions} - if prompts: - record.update({"round": round_no, - "parent": [str(p) for p in prior_paths], - "aliases": aliases}) - path.write_text(json.dumps(record, indent=2)) + # Final write: same file the fan-out has been updating, now complete and with + # the synthesis attached. Clears the `partial` marker. Revision metadata + # (round/parent/aliases) is folded in by save() when this is a revision round. + 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))