fix(pilot): preserve partial results and stop on provider billing errors

docs/experiments/P0-execution-2026-10-28.md

@@ -0,0 +1,82 @@
+# P0 execution report: 2026-10-28
+
+**Status: blocked by provider billing. No GO, PARTIAL or NO-GO decision.**
+The owner requested delivery of the fixes and this report now, with the remaining paid calls deferred.
+
+## What ran
+
+The owner explicitly authorized sending the 65 textbook passages and corresponding Lean code
+through the existing DashScope endpoint to `qwen-max`, at temperature 0 and a cumulative ceiling
+of 400 HTTP requests. The run issued **131 HTTP requests**. Provider responses accounted for
+**37,699 input tokens and 7,790 output tokens**, 45,489 total. These are recorded token counts,
+not an invoice or a monetary cost estimate. The client counted 89 received model responses;
+a received response is not necessarily a usable detector result.
+
+| Coverage | Count |
+|---|---:|
+| Original cohort | 65 |
+| Items with both detectors complete | 42 |
+| Items with one or both detectors missing | 23 |
+| Successful role-extraction/alignment components | 45 |
+| Successful judge components | 43 |
+| Missing role components | 20 |
+| Missing judge components | 22 |
+| Remaining request reservations | 269 |
+
+DashScope returned HTTP 400 with error code `Arrearage` and its overdue-payment reference.
+The first runner implementation recorded the failures and continued through the cohort, so
+65 saved rows did not mean 65 successful items. The scorer rejected these incomplete outputs.
+No partial-sample gate decision or performance comparison has been produced.
+
+## Inputs and labels
+
+The default `items.jsonl` export omitted `source_text`. All 65 original passages were recovered
+from the matching detection spans and checked byte-for-byte against the archived private
+`review.json`. Item IDs and Lean code were unchanged. The recovery tool also verifies every
+exported field against the archive, normalizing the archive's JSON-encoded imports field.
+Source-inclusive inputs and the regenerated labeling HTML remain in the ignored data directory.
+
+The owner replaced the planned Kimi/human labeling with Codex annotation. All 65 labels were
+frozen before the first detector request. They contain 30 faithful, 25 wrong, 7 minor drift and
+3 not-formalizable verdicts. This is one AI annotator, **not human gold**. The annotator saw the
+detector implementation and aggregate compilation counts, but not per-item compiler outcomes
+or detector outputs. Syntax/proof defects were ignored when the intended semantics were recoverable;
+no compiler or API-name audit was performed. Independent human auditing remains necessary before
+using these labels as evidence for paper-level claims.
+
+## Execution repairs
+
+- Empty passages and mismatched frozen inputs/labels are rejected before paid calls.
+- Each HTTP attempt reserves cumulative budget before sending, including transport and JSON retries.
+  Model, endpoint, temperature, input/label/prompt hashes and cumulative token usage are persisted.
+- Billing/authentication errors now stop immediately, preserving partial results.
+- Resume keeps every successful detector component and retries only missing components. Existing
+  rows are replaced atomically rather than duplicated. A budget stop also preserves completed work
+  within the current item. A lock prevents concurrent writers.
+- `--audit-only` reconciles saved metadata without API calls. It confirmed 42 complete items,
+  23 incomplete items, 42 missing components and 131 already-consumed request reservations.
+- Scoring requires complete results and unchanged frozen-label/result hashes. It restores the
+  preregistered complementary-signal PARTIAL rule and uses wrong versus faithful for primary metrics;
+  minor drift is included as positive only in the secondary comparison.
+- Detector prompts, matching thresholds, the concept table and frozen annotations were not tuned
+  against these outputs. Recovery changes concern transport, persistence and accounting only.
+
+The private first-attempt output and metadata were backed up before reconciliation. The results
+hash and the frozen-label hash did not change during the offline audit.
+
+## Validation and next action
+
+149 offline unit tests passed, including billing-stop and partial-resume regressions; repository-wide
+Ruff and diff checks passed. These checks did not compile Lean or build Docker locally.
+The initial PR revision passed Python tests and real Lean validation on
+`ubuntu24.04-x64-lean4.35`, runner 7, in [CI #427](https://gitea.entroia.com/Zayden-Jung/lean-bench/actions/runs/427).
+The image job was skipped by the PR workflow. The recovery revision's remote result is reported separately in the PR checks.
+
+After the owner confirms billing recovery, resume the existing run with the same inputs, frozen
+labels and cumulative budget. There are 42 missing components; retries may consume more requests.
+Do not change model or prompts to work around the provider block. Only after all 65 items have
+successful outputs should the scorer produce the registered decision and disagreement report.
+Phase 1 has not started.
+
+Tracked audit data: `experiments/P0-pilot/results/execution.json` (counts and hashes only).
+Private raw files: `experiments/P0-pilot/data/` (gitignored).

docs/experiments/P0-pilot.md

@@ -2,7 +2,11 @@
 
 Experiment ID **P0**, Phase 0. Registry and conventions: [README.md](README.md).
 
-Status: **approved 2026-10-22; 65 AI labels frozen 2026-10-28; detector execution pending explicit data-transfer approval. No pilot API request has been sent.**
+Status: **approved 2026-10-22; 65 AI labels frozen 2026-10-28; detector run blocked by DashScope Arrearage after 131 requests. 42 items are complete, 23 need missing detector components. No gate decision has been issued.**
+
+The owner explicitly authorized sending the 65 passages and Lean code to DashScope/qwen-max,
+then requested fixes and an execution report now, with paid recovery calls deferred.
+Execution report: [P0 execution, 2026-10-28](P0-execution-2026-10-28.md).
 
 Owner amendment (2026-10-28, in the execution chat): Codex now handles preparation,
 annotation, detector execution and reporting; Kimi's participation is no longer required.

experiments/P0-pilot/README.md

@@ -44,6 +44,9 @@ Labeling must be finished before anyone opens `results.jsonl`.
 - Metadata is stored beside the output as `results.meta.json`. Model, endpoint, temperature,
   input hash, frozen-label hash and prompt hashes must match on resume. Usage is cumulative.
   A file lock prevents concurrent writers. A budget stop returns exit status 2.
+- Billing/authentication failures stop immediately. Resume retries only missing detector components,
+  keeping both complete items and the successful side of partial items. Rows are replaced atomically.
+  `--audit-only` reconciles saved completion counts and status without any API request.
 - The nominal successful path uses 130 calls (one source extraction and one judge per item);
   malformed replies and network retries may increase the actual count.
 - Scoring requires complete labels and results and no unresolved detector errors. The primary
@@ -52,3 +55,11 @@ Labeling must be finished before anyone opens `results.jsonl`.
   by the judge. No threshold, concept table or detector prompt was tuned for these labels.
 - Raw passages, labeling HTML and model responses remain in the ignored `data/` directory.
   Only verdicts, abstract annotation notes, hashes and text-free summaries belong in Git.
+
+## Current execution
+
+The 2026-10-28 run was blocked by DashScope `Arrearage`: 42 items are complete, 23 need
+42 detector components, and 131 of the 400 request reservations have been consumed.
+The owner requested fixes and a report now, with paid resumption deferred.
+See `docs/experiments/P0-execution-2026-10-28.md` and `results/execution.json`.
+No gate decision has been issued. Restore provider billing before resuming the same run.

experiments/P0-pilot/results/execution.json

@@ -0,0 +1,71 @@
+{
+  "experiment": "P0",
+  "status": "provider_blocked",
+  "decision": null,
+  "decision_withheld_reason": "Incomplete detector outputs after DashScope Arrearage; no GO/PARTIAL/NO-GO is valid.",
+  "cohort_size": 65,
+  "completed_items": 42,
+  "items_with_failures": 23,
+  "missing_components": {
+    "judge": 22,
+    "role": 20
+  },
+  "request_attempts": 131,
+  "max_calls": 400,
+  "remaining_request_budget": 269,
+  "usage": {
+    "calls": 89,
+    "prompt_tokens": 37699,
+    "completion_tokens": 7790
+  },
+  "model": "qwen-max",
+  "temperature": 0.0,
+  "items_sha256": "487ecd22919a5e3b0174a2846ab8fe7c0daf4d3feccccf8ea6571aec04c41bed",
+  "labels_sha256": "1552dba85e75426799151384e7e28c9a0b1072ed4c862df0876df89920d2a638",
+  "results_sha256": "fce402e6eab460a48bb9a8e94140bb23f305843ab46d227bbffa5100d7de69e5",
+  "prompt_hashes": {
+    "source": "382f9a7b5a0cfb2d50d525f884dc806f45adaa59bcc45e388920c4759db1546f",
+    "judge": "7ab92682bd71d78a212471289f0f01db39d6b68f49d44df65d7b1c9e34ebb5e2"
+  },
+  "labels": {
+    "type": "single AI annotator; not human gold",
+    "frozen": true,
+    "count": 65
+  },
+  "affected_blind_ids": [
+    "P02",
+    "P04",
+    "P09",
+    "P18",
+    "P19",
+    "P20",
+    "P21",
+    "P22",
+    "P24",
+    "P27",
+    "P29",
+    "P30",
+    "P33",
+    "P36",
+    "P37",
+    "P38",
+    "P42",
+    "P43",
+    "P45",
+    "P47",
+    "P54",
+    "P59",
+    "P62"
+  ],
+  "successful_components_preserved": 88,
+  "resume_policy": "After owner confirms billing recovery, retry missing components only using the same run files, frozen labels, model, prompts and cumulative 400-request ceiling.",
+  "owner_instruction": "2026-10-28: complete fixes and the execution report now; resume paid calls later.",
+  "validation": {
+    "offline_unit_tests": 149,
+    "ruff": "passed",
+    "remote_runner_tag": "ubuntu24.04-x64-lean4.35",
+    "initial_remote_ci": "https://gitea.entroia.com/Zayden-Jung/lean-bench/actions/runs/427",
+    "initial_remote_ci_status": "success; image job skipped for PR",
+    "recovery_remote_ci": "See PR #5 checks for the delivered head revision"
+  }
+}

experiments/P0-pilot/run_pilot.py

@@ -50,37 +50,64 @@ def judge(client, passage: str, lean: str) -> dict:
             "reason": str(reply.get("reason", ""))[:300]}
 
 
-def run_item(client, item: dict) -> dict:
-    if not item.get("source_text", "").strip():
-        raise ValueError("Missing source_text; use a private source-inclusive export")
-    code = item.get("lean_code") or ""
-    row: dict = {"id": item["id"], "compiled": bool(item.get("verified")), "errors": {}}
-    proj = project(code)
-    row["projection"] = proj.as_dict()
-    row["lint"] = lint(proj)
-    try:
-        src = srcsimplex.extract_source_simplex(client, item.get("source_text", ""),
-                                                item.get("label", ""))
-        al = align(src, proj)
-        row["source"] = [r.__dict__ for r in src.roles]
-        row["role"] = al.as_dict()
-    except CallBudgetExceeded:
-        raise
-    except Exception as error:  # a failed extraction is data, not a crash
-        row["errors"]["role"] = str(error)[:300]
-    try:
-        row["judge"] = judge(client, item.get("source_text", ""), code)
-    except CallBudgetExceeded:
-        raise
-    except Exception as error:
-        row["errors"]["judge"] = str(error)[:300]
-    return row
+def provider_blocked(error: Exception) -> bool:
+    text = str(error).lower()
+    return any(code in text for code in ("arrearage", "overdue-payment", "invalid_api_key", "http 401", "http 403"))
+
+
+def complete_row(row: dict) -> bool:
+    return not row.get("errors") and "role" in row and "judge" in row
+
+
+class ProviderBlocked(RuntimeError):
+    """A billing or authentication failure requires external recovery before resuming."""
+
+    def __init__(self, row):
+        super().__init__("Provider billing/authentication blocked; partial results were preserved")
+        self.row = row
 
 
 class CallBudgetExceeded(RuntimeError):
     """Stop before issuing a request that would exceed the cumulative budget."""
 
 
+def run_item(client, item: dict, previous: dict | None = None) -> dict:
+    if not item.get("source_text", "").strip():
+        raise ValueError("Missing source_text; use a private source-inclusive export")
+    code = item.get("lean_code") or ""
+    row: dict = {**(previous or {}), "id": item["id"], "compiled": bool(item.get("verified"))}
+    row["errors"] = dict(row.get("errors", {}))
+    proj = project(code)
+    row["projection"] = proj.as_dict()
+    row["lint"] = lint(proj)
+    if "role" not in row:
+        try:
+            src = srcsimplex.extract_source_simplex(client, item["source_text"], item.get("label", ""))
+            al = align(src, proj)
+            row["source"] = [r.__dict__ for r in src.roles]
+            row["role"] = al.as_dict()
+            row["errors"].pop("role", None)
+        except CallBudgetExceeded as error:
+            error.row = row
+            raise
+        except Exception as error:
+            row["errors"]["role"] = str(error)[:300]
+            if provider_blocked(error):
+                raise ProviderBlocked(row) from error
+    if "judge" not in row:
+        try:
+            row["judge"] = judge(client, item["source_text"], code)
+            row["errors"].pop("judge", None)
+        except CallBudgetExceeded as error:
+            error.row = row
+            raise
+        except Exception as error:
+            row["errors"]["judge"] = str(error)[:300]
+            if provider_blocked(error):
+                raise ProviderBlocked(row) from error
+    return row
+
+
 def digest(path: Path) -> str:
     return hashlib.sha256(path.read_bytes()).hexdigest()
 
@@ -106,8 +133,9 @@ def execute(args, items, settings) -> int:
     if args.out.exists():
         rows = [json.loads(x) for x in args.out.read_text(encoding="utf-8").splitlines() if x.strip()]
     ids = {item["id"] for item in items}
-    done = {row["id"] for row in rows}
-    if len(done) != len(rows) or not done <= ids:
+    by_id = {row["id"]: row for row in rows}
+    done = {row["id"] for row in rows if complete_row(row)}
+    if len(by_id) != len(rows) or not set(by_id) <= ids:
         raise ValueError("Results have duplicate or unknown item IDs")
     hashes = {
         "source": hashlib.sha256((srcsimplex.SYSTEM_PROMPT + inspect.getsource(srcsimplex.passage_prompt)).encode()).hexdigest(),
@@ -129,11 +157,24 @@ def execute(args, items, settings) -> int:
     elif rows:
         raise ValueError("Cannot resume results without their audit metadata")
     todo = [item for item in items if item["id"] not in done]
+    missing = sum(2 - int("role" in by_id.get(i["id"], {}))
+                  - int("judge" in by_id.get(i["id"], {})) for i in todo)
     print(f"items={len(items)} done={len(done)} todo={len(todo)} model={settings.llm.model} "
-          f"nominal_calls={len(todo) * 2} request_attempts={meta['request_attempts']} cap={args.max_calls}")
+          f"missing_components={missing} request_attempts={meta['request_attempts']} cap={args.max_calls}")
     print("prompt_hashes:", json.dumps(hashes))
     if args.dry_run:
         return 0
+    if getattr(args, "audit_only", False):
+        meta["completed_items"] = len(done)
+        meta["processed_items"] = len(rows)
+        meta["status"] = ("provider_blocked" if any(provider_blocked(RuntimeError(message))
+                          for row in rows for message in row.get("errors", {}).values())
+                          else "detector_errors" if todo else "complete")
+        if args.out.exists():
+            meta["results_sha256"] = digest(args.out)
+        atomic_json(meta_path, meta)
+        print(json.dumps(meta))
+        return 0 if meta["status"] == "complete" else 2
     if settings.llm.model != "qwen-max" or settings.llm.temperature != 0:
         raise ValueError("P0 is preregistered for qwen-max at temperature 0")
     client = AuditedClient.from_settings(settings.llm)
@@ -142,6 +183,7 @@ def execute(args, items, settings) -> int:
     def checkpoint():
         meta["usage"] = client.usage.as_dict()
         meta["completed_items"] = len(done)
+        meta["processed_items"] = len(by_id)
         atomic_json(meta_path, meta)
 
     def reserve():
@@ -155,19 +197,30 @@ def execute(args, items, settings) -> int:
     meta["max_calls"] = args.max_calls
     meta["status"] = "running"
     checkpoint()
+    def save_row(row):
+        by_id[row["id"]] = row
+        if complete_row(row):
+            done.add(row["id"])
+        temporary = args.out.with_suffix(args.out.suffix + ".tmp")
+        with temporary.open("w", encoding="utf-8") as stream:
+            for item in items:
+                if item["id"] in by_id:
+                    stream.write(json.dumps(by_id[item["id"]], ensure_ascii=False) + "\n")
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, args.out)
+        checkpoint()
+
     try:
-        with args.out.open("a", encoding="utf-8") as stream:
-            for item in todo:
-                row = run_item(client, item)
-                stream.write(json.dumps(row, ensure_ascii=False) + "\n")
-                stream.flush()
-                os.fsync(stream.fileno())
-                done.add(item["id"])
-                checkpoint()
-                print(f"completed={len(done)}/{len(items)} requests={meta['request_attempts']}", flush=True)
-        meta["status"] = "complete"
-    except CallBudgetExceeded as error:
-        meta["status"] = "budget_exhausted"
+        for item in todo:
+            row = run_item(client, item, by_id.get(item["id"]))
+            save_row(row)
+            print(f"completed={len(done)}/{len(items)} requests={meta['request_attempts']}", flush=True)
+        meta["status"] = "complete" if len(done) == len(items) else "detector_errors"
+    except (CallBudgetExceeded, ProviderBlocked) as error:
+        if getattr(error, "row", None) is not None:
+            save_row(error.row)
+        meta["status"] = "provider_blocked" if isinstance(error, ProviderBlocked) else "budget_exhausted"
         print(str(error), file=sys.stderr)
     except BaseException:
         meta["status"] = "interrupted"
@@ -186,6 +239,7 @@ def main() -> int:
     ap.add_argument("--out", required=True, type=Path)
     ap.add_argument("--max-calls", type=int, default=400)
     ap.add_argument("--dry-run", action="store_true")
+    ap.add_argument("--audit-only", action="store_true", help="Reconcile saved state without making API calls")
     ap.add_argument("--labels", type=Path, help="Frozen labels; their hash is bound to the run")
     args = ap.parse_args()
     if not 1 <= args.max_calls <= 400:

tests/test_pilot_scripts.py

@@ -141,3 +141,82 @@ def test_judge_rejects_string_boolean():
     rp = load("run_pilot")
     with pytest.raises(ValueError, match="unusable judge"):
         rp.judge(ScriptedClient(['{"faithful": "false"}']), "passage", "code")
+
+
+def test_partial_resume_keeps_successful_detector_output():
+    rp = load("run_pilot")
+    row = rp.run_item(Stub(), ITEM)
+    expected_role = row["role"]
+    row.pop("judge")
+    row["errors"]["judge"] = "temporary failure"
+
+    class JudgeOnly:
+        calls = 0
+
+        def complete_json(self, messages):
+            self.calls += 1
+            assert "faithful" in messages[0]["content"]
+            return {"faithful": True}
+
+    client = JudgeOnly()
+    resumed = rp.run_item(client, ITEM, row)
+    assert client.calls == 1
+    assert resumed["role"] == expected_role
+    assert rp.complete_row(resumed)
+    assert row["errors"] == {"judge": "temporary failure"}
+
+
+def test_billing_failure_stops_before_next_detector():
+    import pytest
+
+    from leanbench.llm import LLMError
+
+    rp = load("run_pilot")
+
+    class Blocked:
+        calls = 0
+
+        def complete_json(self, messages):
+            self.calls += 1
+            raise LLMError("HTTP 400: Arrearage overdue-payment")
+
+    client = Blocked()
+    with pytest.raises(rp.ProviderBlocked) as caught:
+        rp.run_item(client, ITEM)
+    assert client.calls == 1
+    assert "role" in caught.value.row["errors"]
+    assert "judge" not in caught.value.row["errors"]
+
+
+def test_blocked_run_preserves_and_replaces_partial_row(tmp_path, monkeypatch):
+    import argparse
+    import json
+
+    from leanbench.config import Settings
+
+    rp = load("run_pilot")
+    args = argparse.Namespace(items=tmp_path / "items.jsonl", out=tmp_path / "results.jsonl",
+                              max_calls=400, dry_run=False)
+    args.items.write_text(json.dumps(ITEM) + "\n")
+    partial = {"id": "x1", "compiled": True, "role": {"flagged": False},
+               "errors": {"judge": "Arrearage"}}
+
+    def blocked(client, item, previous):
+        raise rp.ProviderBlocked(partial)
+
+    monkeypatch.setattr(rp, "run_item", blocked)
+    assert rp.execute(args, [ITEM], Settings()) == 2
+    meta = json.loads(args.out.with_suffix(".meta.json").read_text())
+    assert meta["status"] == "provider_blocked"
+    assert meta["completed_items"] == 0
+
+    def recover(client, item, previous):
+        assert previous["role"] == partial["role"]
+        return {**previous, "judge": {"faithful": True}, "errors": {}}
+
+    monkeypatch.setattr(rp, "run_item", recover)
+    assert rp.execute(args, [ITEM], Settings()) == 0
+    rows = [json.loads(line) for line in args.out.read_text().splitlines()]
+    assert len(rows) == 1 and rp.complete_row(rows[0])
+    meta = json.loads(args.out.with_suffix(".meta.json").read_text())
+    assert meta["completed_items"] == 1 and meta["status"] == "complete"