Preserve score-all-options cache across lanes

Co-Authored-By: PI[openai-codex] <288921227+claudypoo@users.noreply.github.com>

scripts/wvs_map.py

@@ -46,6 +46,7 @@ from moralmaps.zones import zones_for, zone_of, IW_MACRO
 from moralmaps.instrument import Instrument, InstrItem
 from moralmaps.read import read_items, resolve_answer_ids
 from moralmaps.read_api import rated_protocol_identity, read_items_rated
+from moralmaps.rated_cache import merge_completed
 from moralmaps.iw_axes import AXIS_ITEMS, X_AXIS, Y_AXIS, SKIP, resolve_items, positiveness
 
 # option labels are single digits 0..n-1 -- single-token (unlike '10' on the justifiable scale) and
@@ -326,10 +327,9 @@ def main() -> None:
                 models[entry["display_key"]] = tuple(entry["coords"])
 
     def save_cache() -> None:
-        """Atomic cache replacement after a complete model panel, so interruption cannot fabricate a hit."""
-        temp = cpath.with_suffix(cpath.suffix + ".tmp")
-        temp.write_text(json.dumps(cache, indent=2, sort_keys=True) + "\n")
-        temp.replace(cpath)
+        """Merge completed panels under a lock so parallel lanes cannot erase one another."""
+        nonlocal cache
+        cache = merge_completed(cpath, cache["completed"])
 
     rng = np.random.default_rng(0)                       # deterministic bootstrap
 
@@ -359,8 +359,7 @@ def main() -> None:
             m, rated_items, n_samples=args.api_samples, temperature=1.0,
             max_tokens=args.api_max_tokens, concurrency=args.api_concurrency,
             req_timeout=args.api_request_timeout, reasoning=api_reasoning,
-            structured_output=args.api_structured_output, provider=api_provider,
-            probe_first=args.api_probe_first)
+            structured_output=args.api_structured_output, provider=api_provider)
         completed = cache["completed"].get(protocol_id)
         if completed is not None:
             models[key] = tuple(completed["coords"])

scripts/wvs_score_all_options_recover_cache.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+"""Recover complete score-all-options panels from the durable request ledger."""
+from __future__ import annotations
+
+import argparse
+import json
+import multiprocessing
+import tempfile
+from collections import defaultdict
+from pathlib import Path
+
+import numpy as np
+
+from moralmaps.iw_axes import X_AXIS, Y_AXIS, resolve_items
+from moralmaps.rated_cache import merge_completed
+from wvs_map import load_wvs_all, model_coord_ci
+
+CACHE = Path("slop/research/wvs/20260916_openrouter/wvs_iw_rated.json")
+RECORDS = Path("slop/research/wvs/20260916_openrouter/wvs_iw_requests.jsonl")
+AUDIT = Path("slop/audits/20260917_wvs_score_all_options_cache_recovery.json")
+
+
+def read_records() -> list[dict]:
+    records = []
+    for line_number, line in enumerate(RECORDS.read_text().splitlines(), start=1):
+        try:
+            records.append(json.loads(line))
+        except json.JSONDecodeError as error:
+            raise ValueError(f"invalid JSONL record at {RECORDS}:{line_number}") from error
+    return records
+
+
+def recovered_entries(records: list[dict]) -> dict[str, dict]:
+    starts = {record["run_id"]: record for record in records if record.get("event") == "run_started"}
+    finished = [record for record in records if record.get("event") == "run_finished"
+                and record["planned_requests"] == 144 and record["valid_samples"] == 144
+                and record["failed_samples"] == 0]
+    item_results: dict[str, dict[str, dict]] = defaultdict(dict)
+    for record in records:
+        if record.get("event") == "item_result":
+            item_results[record["run_id"]][record["id"]] = record
+    resolved = resolve_items(load_wvs_all())
+    expected_ids = [item["suffix"] for axis in (X_AXIS, Y_AXIS) for item in resolved[axis]]
+    entries = {}
+    rng = np.random.default_rng(0)
+    for finish in finished:
+        run_id = finish["run_id"]
+        start = starts[run_id]
+        rows = item_results[run_id]
+        if set(rows) != set(expected_ids):
+            raise ValueError(f"complete run {run_id} has item results {sorted(rows)}, expected {expected_ids}")
+        if any(row["valid_samples"] != 12 for row in rows.values()):
+            raise ValueError(f"complete run {run_id} has non-12 item samples")
+        psamples = {item_id: np.asarray(rows[item_id]["p_samples"]) for item_id in expected_ids}
+        coords = model_coord_ci(psamples, resolved, rng)
+        model = finish["model"]
+        entries[finish["protocol_id"]] = {
+            "model": model,
+            "display_key": model.split("/")[-1] + " (rated)",
+            "coords": list(coords),
+            "records_path": str(RECORDS),
+            "run_id": run_id,
+            "protocol_id": finish["protocol_id"],
+            "n_items": len(rows),
+            "n_samples": 12,
+        }
+    return entries
+
+
+def _writer(path: str, key: str) -> None:
+    merge_completed(Path(path), {key: {"model": key}})
+
+
+def concurrency_smoke() -> None:
+    with tempfile.TemporaryDirectory() as directory:
+        path = Path(directory) / "cache.json"
+        processes = [multiprocessing.Process(target=_writer, args=(str(path), key)) for key in ("a", "b")]
+        for process in processes:
+            process.start()
+        for process in processes:
+            process.join()
+            if process.exitcode != 0:
+                raise RuntimeError(f"cache writer exited {process.exitcode}")
+        assert set(json.loads(path.read_text())["completed"]) == {"a", "b"}
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--smoke", action="store_true")
+    args = parser.parse_args()
+    if args.smoke:
+        concurrency_smoke()
+        print("smoke: two concurrent cache writers preserve both completed entries")
+    records = read_records()
+    entries = recovered_entries(records)
+    merged = merge_completed(CACHE, entries)
+    audit = {
+        "ledger": str(RECORDS),
+        "ledger_valid_lines": len(records),
+        "recovered_complete_runs": len(entries),
+        "cache_completed_entries_after_merge": len(merged["completed"]),
+        "recovered_models": sorted(entry["model"] for entry in entries.values()),
+    }
+    AUDIT.write_text(json.dumps(audit, indent=2, sort_keys=True) + "\n")
+    print(json.dumps(audit, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()