Persist each lane's answer as it lands, not after the last one (#58)

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.

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))