diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 8d285564..43ce9370 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -69,6 +69,7 @@ jobs:
             tests/test_refusal_question_scoped_coverage.py \
             tests/test_refusal_v5_3_preparation.py \
             tests/test_refusal_prospective_preparation.py \
+            tests/test_refusal_prospective_annotation.py \
             tests/test_refusal_v5_3_runner.py \
             tests/test_refusal_v5_2_1_preparation.py \
             tests/test_refusal_v5_2_1_runner.py \
diff --git a/experiments/evals/refusal_prospective_annotation.py b/experiments/evals/refusal_prospective_annotation.py
new file mode 100644
index 00000000..74cd11eb
--- /dev/null
+++ b/experiments/evals/refusal_prospective_annotation.py
@@ -0,0 +1,163 @@
+"""PR80 annotation readiness and preprediction AI review; explicit approval required."""
+
+import argparse
+import hashlib
+import importlib.util
+import json
+from collections import Counter
+from pathlib import Path
+
+from experiments.evals import refusal_prospective_preparation as prep
+
+shared, canonical = prep.shared, prep.canonical
+DATA, ROOT = prep.DATA, prep.DATA / "blind_annotation"
+CONTRACT = prep.selection.prior.REPORT_DIR / "prospective_annotation_contract.json"
+FREEZE = prep.selection.prior.REPORT_DIR / "prospective_annotation_freeze.json"
+PREPARATION_SHA256 = "e093d72579836ecc13f10d64e63a2e62e2a76dbd3bd4b193696427de61edb8b8"
+AUTHORIZATION = "Requires explicit user approval of PR80: 50 full-Top14 Singapore qwen3.6-plus-2026-04-02 annotation + 100 qwen3.5-plus-2026-04-20 paired classification calls / CNY25; annotation cap50/CNY10, classifier cap100/CNY15, no paid retries, DEV, recharge or promotion. A merge-only/continue reply does not establish this approval."
+
+# ponytail: use the frozen durable annotator in an isolated namespace, no new client.
+_spec = importlib.util.spec_from_file_location("_prospective_annotation_engine", prep.annotation.__file__)
+_engine = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_engine)
+
+
+def load_inputs():
+    manifest = shared._read_json(prep.REPORT)
+    if canonical(manifest) != PREPARATION_SHA256:
+        raise RuntimeError("frozen preparation changed")
+    for name, digest in manifest["source_hashes"].items():
+        if hashlib.sha256((Path("experiments/evals") / name).read_text(encoding="utf-8").encode()).hexdigest() != digest:
+            raise RuntimeError("frozen source changed")
+    if hashlib.sha256(prep.PROTOCOL.read_text(encoding="utf-8").encode()).hexdigest() != manifest["protocol_normalized_sha256"]:
+        raise RuntimeError("frozen protocol changed")
+    rows = prep.paired.prior.paired.v2.load_classifier_inputs(DATA / "inputs.jsonl", expected_sha256=manifest["inputs_sha256"], expected_cases=50)
+    if [row["question_id"] for row in rows] != manifest["selected_question_ids"]:
+        raise RuntimeError("frozen order changed")
+    arm = manifest["preflight"]["arms"]["annotation"]
+    for index, row in enumerate(rows):
+        messages = prep.annotation.messages(row)
+        if (canonical(messages) != arm["request_sha256"][index]
+                or sum(len(m["content"].encode()) for m in messages) + 512 != arm["input_byte_plus_framing_bounds"][index]):
+            raise RuntimeError("annotation request changed")
+    return manifest, rows
+
+
+def execution_contract(manifest):
+    arm = manifest["preflight"]["arms"]["annotation"]
+    return {"run": "prospective_refusal_annotation_v1", "preparation_canonical_sha256": canonical(manifest),
+        "model": prep.annotation.MODEL, "region": "Singapore", "cases": 50, "max_calls": 50,
+        "max_output_tokens": 4096, "hard_accounting_cap_cny": 10,
+        "input_cny_per_million": 3.75, "output_cny_per_million": 22.5,
+        "inputs_sha256": manifest["inputs_sha256"], "prompt_sha256": arm["prompt_sha256"],
+        "request_sha256": arm["request_sha256"], "input_byte_bounds": arm["input_byte_plus_framing_bounds"],
+        "runner_normalized_source_sha256": hashlib.sha256(Path(__file__).read_text(encoding="utf-8").encode()).hexdigest(),
+        "pricing": manifest["preflight"]["pricing"], "planning_cost_cny": arm["planning_cost_cny"],
+        "authorization": AUTHORIZATION, "controls": {"sdk_retries": 0, "paid_retries": False,
+            "raw_before_parse": True, "unknown_charge_blocks_resume": True, "dev_opened": False,
+            "promotion_allowed": False, "review_and_target_freeze_before_classifier": True}}
+
+
+def prepare_contract(*, authorization_confirmed=False):
+    if authorization_confirmed is not True:
+        raise RuntimeError("explicit PR80 data/model/budget authorization required")
+    manifest, _ = load_inputs()
+    value = execution_contract(manifest)
+    prep.selection.prior._write_json(CONTRACT, value)
+    return value
+
+
+def load_probe():
+    manifest, rows = load_inputs()
+    contract = shared._read_json(CONTRACT)
+    if contract != execution_contract(manifest):
+        raise RuntimeError("frozen annotation contract changed")
+    return contract, rows
+
+
+def run_annotation(*, root=ROOT, client=None, max_new_calls=50, probe_loader=load_probe):
+    if type(max_new_calls) is not int or not 0 <= max_new_calls <= 50:
+        raise ValueError("max_new_calls must be between 0 and 50")
+    contract, rows = probe_loader()
+    # No resume-after-audit switch: only a validated saved-response recovery can advance a failed row.
+    return _engine.run(contract, rows, root, client=client, max_new_calls=max_new_calls)
+
+
+def accounted_draft(*, root=ROOT):
+    state = run_annotation(root=root, max_new_calls=0)
+    if state["completed"] != 50 or state["calls"] != 50 or (Path(root) / "runner.lock").exists():
+        raise RuntimeError("all50 stopped, accounted annotation outcomes required")
+    _, rows = load_probe()
+    events = shared._read_jsonl(Path(root) / "attempts.jsonl")
+    path = Path(root) / "citation_recoveries.jsonl"
+    recoveries = {r["call_index"]: r["annotation"] for r in shared._read_jsonl(path)} if path.exists() else {}
+    draft = [{"question_id": row["question_id"], "annotation": recoveries[index] if index in recoveries else event["result"]["annotation"]}
+             for index, (row, event) in enumerate(zip(rows, events, strict=True))]
+    return state, events, rows, draft
+
+
+def export_draft():
+    _, _, _, draft = accounted_draft()
+    prep.selection.prior._write_jsonl(DATA / "blind_draft_annotations.jsonl", draft)
+    return len(draft)
+
+
+def freeze_review(review_path, *, root=ROOT):
+    if (DATA / "paired_run_v1").exists():
+        raise RuntimeError("classifier checkpoint exists before review freeze")
+    state, events, rows, draft = accounted_draft(root=root)
+    if shared._read_jsonl(DATA / "blind_draft_annotations.jsonl") != draft:
+        raise RuntimeError("frozen draft differs from accounted raw responses")
+    review = shared._read_jsonl(review_path)
+    changed = prep.selection.validate_review(rows, draft, review)
+    targets = [{"question_id": r["question_id"], "target_class": r["annotation"]["target_class"]} for r in review]
+    path = DATA / "targets.jsonl"
+    if FREEZE.exists() or path.exists() or (DATA / "reviewed_annotations.jsonl").exists():
+        raise RuntimeError("refusing to overwrite target/review freeze")
+    prep.selection.prior._write_jsonl(DATA / "reviewed_annotations.jsonl", review)
+    prep.selection.prior._write_jsonl(path, targets)
+    counts = Counter(t["target_class"] for t in targets)
+    recovery_path = Path(root) / "citation_recoveries.jsonl"
+    value = {"status": "AI_PROXY_TARGETS_FROZEN_BEFORE_CLASSIFIER", "cases": 50,
+        "counts": {c: counts[c] for c in ("SUFFICIENT", "INSUFFICIENT", "AMBIGUOUS")},
+        "class_changes_from_draft": changed, "provider_calls": len(events),
+        "estimated_annotation_cost_cny": state["estimated_cost_cny"],
+        "prompt_tokens": sum(e["prompt_tokens"] for e in events), "completion_tokens": sum(e["completion_tokens"] for e in events),
+        "original_schema_failures": sum(e["result"] is None for e in events),
+        "contract_canonical_sha256": canonical(shared._read_json(CONTRACT)),
+        "inputs_sha256": shared._sha256(DATA / "inputs.jsonl"), "targets_sha256": shared._sha256(path),
+        "review_sha256": shared._sha256(DATA / "reviewed_annotations.jsonl"),
+        "draft_sha256": shared._sha256(DATA / "blind_draft_annotations.jsonl"),
+        "journal_sha256": shared._sha256(Path(root) / "attempts.jsonl"),
+        "citation_recoveries_sha256": shared._sha256(recovery_path) if recovery_path.exists() else None,
+        "review_depth": "Current assistant reviewed every question/draft and cited evidence with targeted fuller context; not an independent full-source expert adjudication",
+        "class_support_at_least_five_each": all(counts[c] >= 5 for c in ("SUFFICIENT", "INSUFFICIENT")),
+        "classifier_predictions_created": False, "dev_opened": False, "promotion_allowed": False,
+        "provenance": "Same-family Qwen draft plus current-assistant preprediction review: AI proxy, not independent human/expert gold; all TRAIN previously used in retrieval/analysis"}
+    prep.selection.prior._write_json(FREEZE, value)
+    return value
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    mode = parser.add_mutually_exclusive_group(required=True)
+    mode.add_argument("--prepare-contract", action="store_true")
+    mode.add_argument("--run-paid", action="store_true")
+    mode.add_argument("--export-draft", action="store_true")
+    mode.add_argument("--freeze-review")
+    parser.add_argument("--env-file")
+    parser.add_argument("--max-new-calls", type=int, default=50)
+    parser.add_argument("--confirm-pr80-authorization", action="store_true",
+                        help="Record an already-obtained explicit user approval; this flag is not permission itself")
+    args = parser.parse_args()
+    if args.prepare_contract:
+        print(json.dumps({"max_calls": prepare_contract(authorization_confirmed=args.confirm_pr80_authorization)["max_calls"]}))
+    elif args.run_paid:
+        if args.env_file:
+            from dotenv import load_dotenv
+            load_dotenv(args.env_file, override=False)
+        print(json.dumps(run_annotation(max_new_calls=args.max_new_calls)))
+    elif args.export_draft:
+        print(json.dumps({"exported": export_draft()}))
+    else:
+        print(json.dumps(freeze_review(args.freeze_review)))
diff --git a/tests/test_refusal_prospective_annotation.py b/tests/test_refusal_prospective_annotation.py
new file mode 100644
index 00000000..33912e1f
--- /dev/null
+++ b/tests/test_refusal_prospective_annotation.py
@@ -0,0 +1,103 @@
+import hashlib
+import json
+
+import pytest
+
+from experiments.evals import refusal_prospective_annotation as run
+from tests.test_refusal_v5_2_annotation import Client, setup_run
+
+
+def test_preparing_execution_requires_explicit_authorization_before_loading_inputs(monkeypatch):
+    monkeypatch.setattr(run, "load_inputs", lambda: pytest.fail("inputs read without authorization"))
+    for value in (False, None, "continue", 1):
+        with pytest.raises(RuntimeError, match="explicit PR80"):
+            run.prepare_contract(authorization_confirmed=value)
+
+
+def test_resume_recovery_retains_raw_without_paid_retry(tmp_path):
+    contract, rows, value = setup_run()
+    contract["max_calls"] = 3
+    def loader():
+        return contract, rows
+    value["annotation"]["notes"] = "Source 1 (Line 1) supports this fact."
+    bad = Client(raw=json.dumps(value))
+    with pytest.raises(RuntimeError, match="failure saved"):
+        run.run_annotation(root=tmp_path, client=bad, probe_loader=loader)
+    with pytest.raises(RuntimeError, match="resume-after-audit"):
+        run.run_annotation(root=tmp_path, client=bad, probe_loader=loader)
+    assert bad.calls == 1
+    before = (tmp_path / "attempts.jsonl").read_bytes()
+    recovery = {"call_index": 0, "raw_sha256": hashlib.sha256(bad.raw.encode()).hexdigest(),
+                "annotation": value["annotation"] | {"notes": "Source 1:L1 supports this fact."}}
+    run.prep.selection.prior._write_jsonl(tmp_path / "citation_recoveries.jsonl", [recovery])
+    good = Client()
+    assert run.run_annotation(root=tmp_path, client=good, probe_loader=loader)["completed"] == 3
+    assert run.run_annotation(root=tmp_path, client=good, probe_loader=loader)["calls"] == 3
+    assert good.calls == 2 and (tmp_path / "attempts.jsonl").read_bytes().startswith(before)
+
+
+def test_unknown_charge_lock_budget_and_call_caps_block(tmp_path):
+    contract, rows, _ = setup_run()
+    def loader():
+        return contract, rows
+    unknown = Client(fail=True)
+    with pytest.raises(RuntimeError, match="failure saved"):
+        run.run_annotation(root=tmp_path / "unknown", client=unknown, probe_loader=loader)
+    with pytest.raises(RuntimeError, match="unknown charge"):
+        run.run_annotation(root=tmp_path / "unknown", client=unknown, probe_loader=loader)
+    assert unknown.calls == 1
+    good = Client()
+    contract["hard_accounting_cap_cny"] = 0
+    with pytest.raises(RuntimeError, match="cap reached"):
+        run.run_annotation(root=tmp_path / "budget", client=good, probe_loader=loader)
+    (tmp_path / "budget" / "runner.lock").touch()
+    with pytest.raises(FileExistsError):
+        run.run_annotation(root=tmp_path / "budget", client=good, probe_loader=loader)
+    for cap in (True, -1, 51):
+        with pytest.raises(ValueError):
+            run.run_annotation(root=tmp_path, max_new_calls=cap)
+    assert good.calls == 0
+
+
+def test_contract_and_preparation_tamper_fail_before_any_paid_request(tmp_path, monkeypatch):
+    manifest = run.shared._read_json(run.prep.REPORT)
+    value = run.execution_contract(manifest)
+    assert value["max_calls"] == 50 and value["hard_accounting_cap_cny"] == 10
+    assert value["controls"]["paid_retries"] is False and value["controls"]["sdk_retries"] == 0
+    assert run.canonical(manifest) == run.PREPARATION_SHA256
+    path = tmp_path / "tampered.json"
+    path.write_text("{}", encoding="utf-8")
+    monkeypatch.setattr(run.prep, "REPORT", path)
+    monkeypatch.setattr(run.shared, "_read_jsonl", lambda path: pytest.fail("private rows read"))
+    with pytest.raises(RuntimeError, match="preparation changed"):
+        run.load_inputs()
+
+
+def test_pre_prediction_review_freeze_is_complete_and_never_forces_binary(tmp_path, monkeypatch):
+    _, short, value = setup_run()
+    rows = [{**short[0], "question_id": f"TRAIN_Q{i:03}"} for i in range(50)]
+    draft = [{"question_id": row["question_id"], **value} for row in rows]
+    review = [r | {"annotation": r["annotation"] | {"target_class": "AMBIGUOUS"}} for r in draft]
+    root = tmp_path / "blind_annotation"
+    root.mkdir()
+    events = [{"prompt_tokens": 100, "completion_tokens": 100, "result": draft[i]} for i in range(50)]
+    run.prep.selection.prior._write_jsonl(root / "attempts.jsonl", events)
+    for name, values in (("blind_draft_annotations.jsonl", draft), ("review_working.jsonl", review), ("inputs.jsonl", rows)):
+        run.prep.selection.prior._write_jsonl(tmp_path / name, values)
+    contract = tmp_path / "contract.json"
+    run.prep.selection.prior._write_json(contract, {"synthetic": True})
+    monkeypatch.setattr(run, "DATA", tmp_path)
+    monkeypatch.setattr(run, "FREEZE", tmp_path / "freeze.json")
+    monkeypatch.setattr(run, "CONTRACT", contract)
+    monkeypatch.setattr(run, "accounted_draft", lambda **kw: ({"estimated_cost_cny": .1}, events, rows, draft))
+    (tmp_path / "paired_run_v1").mkdir()
+    with pytest.raises(RuntimeError, match="classifier checkpoint"):
+        run.freeze_review(tmp_path / "review_working.jsonl", root=root)
+    (tmp_path / "paired_run_v1").rmdir()
+    result = run.freeze_review(tmp_path / "review_working.jsonl", root=root)
+    assert result["counts"] == {"SUFFICIENT": 0, "INSUFFICIENT": 0, "AMBIGUOUS": 50}
+    assert result["class_changes_from_draft"] == 50 and not result["classifier_predictions_created"]
+    assert not result["class_support_at_least_five_each"] and not result["promotion_allowed"]
+    assert result["targets_sha256"] == run.shared._sha256(tmp_path / "targets.jsonl")
+    with pytest.raises(RuntimeError, match="overwrite"):
+        run.freeze_review(tmp_path / "review_working.jsonl", root=root)