Restore original GAP prompts and lock prompt bytes

18 files changed, 771 insertions, 1366 deletions

diff --git a/GAP_End_to_End.ipynb b/GAP_End_to_End.ipynb
index 92910bb..bdb519f 100644
--- a/GAP_End_to_End.ipynb
+++ b/GAP_End_to_End.ipynb
@@ -38,6 +38,11 @@
     "src_path = str(PACKAGE_ROOT / \"src\")\n",
     "if src_path not in sys.path:\n",
     "    sys.path.insert(0, src_path)\n",
+    "subprocess.run(\n",
+    "    [sys.executable, \"-m\", \"pytest\", \"-q\", str(PACKAGE_ROOT / \"tests\" / \"test_prompts.py\")],\n",
+    "    cwd=PACKAGE_ROOT,\n",
+    "    check=True,\n",
+    ")\n",
     "print(f\"GAP package root: {PACKAGE_ROOT}\")"
    ]
   },
diff --git a/PROMPT_SHA256SUMS b/PROMPT_SHA256SUMS
new file mode 100644
index 0000000..01aa81a
--- /dev/null
+++ b/PROMPT_SHA256SUMS
@@ -0,0 +1,18 @@
+# SHA-256 of UTF-8 prompt values in src/gap_pipeline/prompts.py.
+# These values are pinned to the recovered original source files.
+0c817ede22fd407411ee588a02cf7a14ddcbb3d607aa4db6b5f81619609bcfef  KERNEL_PLAN_SYSTEM
+4aafd1c25d67c5a433c8ad48dcf347cfe53436e81cfde1883e044ca3c17d855d  KERNEL_PLAN_PROMPT
+db0c902e15c9830a3c3860509bdb8a17b412d6b646b0f7e3278f32669046212b  KERNEL_GENERATE_SYSTEM
+89c8a05dcda1b866b7835167c41184de9176ea058922e2cb250325825f79db92  KERNEL_GENERATE_PROMPT
+50d5ee3ccbdf5196afea1eea60e61e74a324c89f14086b0ec6a8e14cb4ad0f30  JUDGE_SYSTEM_PROMPT
+59338f0a2522bda73006e7f62243578596cb8113da90fd8bfe5767d5a2bf4c2b  JUDGE_USER_TEMPLATE
+1c028afeac4bfbf92dd56cfce7b004b0823ab721f6f71e776bb90a5c652c1ed8  FIX_SYSTEM_PROMPT
+f12126acdc1518e872cd61f6baf2eedd49f1ad85eca69438e78128293959a8e9  FIX_USER_TEMPLATE
+d05ab24b4afd4ccd23e7954fa1111ce9c74893f8572c62617a3e568e3a743994  SURFACE_SYSTEM_BASE
+1cca2d06215207aa85ae36b368298a0b13ad4c2ef475079f1df0b2c4d9a5edd1  SURFACE_TASK_COMMON
+06b9b9c6a65ceae6818f30387cfd6e18bb46bbc1ccdf79adbba73f8849f6b354  SURFACE_TASK_DESCRIPTIVE
+7df3748ab186bbb6d9f421de5a7200754105154036d5cf6fb21ee32ac1f61848  SURFACE_TASK_CONFUSING
+fcde08ded42df91b44cb403f9a439550077245d71586adb6da9fb4983a4ea76e  SURFACE_TASK_MISLEADING
+576301ca8e9eff2e4a2d9ed6a74f884a5a881ba22fe3a38e22e5eb46e341fac0  SURFACE_TASK_GARBLED
+b9f33abc9c8ab5e30b0154a6eebbd139d91afea6a744b454d4e150a6f7224a26  SURFACE_RETURN_SPEC
+5d9043d7d1c1033db1aea0696812a2c4cbb6f3e28ca9d7b6436c036b40831fe7  SURFACE_USER_TEMPLATE
diff --git a/README.md b/README.md
index 56b76ec..090c297 100644
--- a/README.md
+++ b/README.md
@@ -1,35 +1,47 @@
 # GAP minimal reproduction package
 
-This repository contains the generation and verification pipeline for GAP:
-four surface-renaming families and one proof-plan-preserving kernel variant.
-It is intentionally small and includes one canonical Putnam example for an
-end-to-end check.
+This repository provides a one-item end-to-end reproduction of GAP: four
+surface-renaming families and one verified kernel variant.
 
 ## One-click reproduction
 
 Open `GAP_End_to_End.ipynb` and choose **Run All**. The notebook:
 
 1. installs the package;
-2. loads one canonical problem;
-3. generates four surface variants;
-4. runs the five-stage kernel pipeline;
-5. applies five judges until the same candidate receives two consecutive
+2. loads one canonical Putnam problem;
+3. generates the four surface variants;
+4. extracts the kernel proof plan and mutable slots;
+5. generates a new question and complete solution with the same plan;
+6. runs five judges until the unchanged candidate receives two consecutive
    unanimous rounds;
-6. exports and validates one machine-readable GAP record.
+7. exports and validates one machine-readable GAP record.
 
-The default live model is `o3`. Set `OPENAI_API_KEY` before starting Jupyter,
-or enter it in the notebook's hidden prompt. The key is not written to the
-notebook or run artifacts.
+The default model is `o3`. Set `OPENAI_API_KEY` before starting Jupyter, or
+enter it in the notebook's hidden prompt. The key is never saved.
 
-To test the complete software path without API calls:
+For a no-API software check:
 
 ```bash
 GAP_OFFLINE_SMOKE=1 jupyter notebook GAP_End_to_End.ipynb
 ```
 
-Offline smoke mode uses scripted model responses. It validates pipeline wiring,
-verification control flow, and release assembly; it does not evaluate generated
-mathematical content.
+Offline smoke mode uses fixed responses. It validates pipeline wiring,
+verification control flow, and release assembly; it does not assess generated
+mathematics.
+
+## Prompt fidelity
+
+Generation, surface-renaming, judge, and repair prompts are copied verbatim
+from the original author source and the prompt listing in the paper. Their
+UTF-8 SHA-256 digests are pinned in `PROMPT_SHA256SUMS` and enforced by
+`tests/test_prompts.py`.
+
+The OpenAI adapter does not send `temperature`; this is compatible with `o3`,
+whose supported value is its default.
+
+The conceptual five stages are represented explicitly in saved artifacts.
+The original implementation batches stages 1–3 into Prompt-A and stages 4–5
+into Prompt-B; the wrapper does not change those prompts. See `STAGE_MAP.md`.
 
 ## Install and test
 
@@ -69,7 +81,7 @@ PYTHONPATH=src python -m gap_pipeline.cli generate-surfaces \
   --proposer-model o3
 ```
 
-Assemble the final record after both runs complete:
+Assemble the final record:
 
 ```bash
 PYTHONPATH=src python -m gap_pipeline.cli export-release \
@@ -80,26 +92,10 @@ PYTHONPATH=src python -m gap_pipeline.cli export-release \
   --item-id 1998-B-1
 ```
 
-## Protocol
-
-The kernel path consists of:
-
-1. reference solution → proof DAG;
-2. proof DAG → content-free method plan;
-3. guarded replacement at a DAG leaf;
-4. replacement diffusion through the fixed DAG;
-5. regenerated proof → self-contained problem statement.
-
-Verification uses `J=5` judges, requires `K=2` consecutive unanimous rounds for
-the same candidate, and permits at most `T=15` rounds. A non-unanimous round
-resets the pass streak and triggers repair.
-
-`STAGE_MAP.md` maps each paper operation to the implementation and its validated
-output. Every model call and intermediate stage is saved below the selected run
-directory for inspection.
-
-## Reproducibility scope
+## Verification protocol
 
-LLM generation is stochastic, so repeated live runs need not produce identical
-wording. This package reproduces the prompts, transformation stages, typed
-structural checks, verification state machine, and release format.
+Kernel verification uses `J=5` judges, requires `K=2` consecutive unanimous
+rounds for the same candidate, and allows at most `T=15` rounds. A rejected
+round resets the streak and triggers a complete question-and-solution repair.
+Every call, stage output, iteration, and final record is saved under the chosen
+run directory.
diff --git a/STAGE_MAP.md b/STAGE_MAP.md
index 1cf0691..26077a3 100644
--- a/STAGE_MAP.md
+++ b/STAGE_MAP.md
@@ -1,17 +1,22 @@
 # GAP paper-to-code map
 
-| Paper operation | Implementation | Validated output |
+The original implementation batches adjacent conceptual stages into two model
+calls. Prompt-A returns the ordered proof-plan nodes and mutable slots; Prompt-B
+re-instantiates that plan and returns the regenerated proof and question.
+The package writes each conceptual output separately so every paper stage is
+visible in an end-to-end run without changing either prompt.
+
+| Paper operation | Implementation | Saved artifact |
 |---|---|---|
-| Surface rename | `surface.py` | `SurfaceVariant` with a collision-free one-to-one identifier map |
-| 1. Reference solution to proof DAG | `KernelPipeline.construct_dag` | Acyclic `ProofDAG` connected to the terminal claim |
-| 2. Content-free method plan | `KernelPipeline.summarize_methods` | `MethodPlan` with the DAG's node IDs and topological order |
-| 3. Guarded leaf replacement | `KernelPipeline.generate_replacement` | `ReplacementSpec` with explicit guards and evidence |
-| 4. Diffusion through the DAG | `KernelPipeline.diffuse_dag` | `DiffusedProof` preserving node order, dependencies, and method labels |
-| 5. Problem rendering | `KernelPipeline.render_question` | Self-contained `KernelCandidate` aligned with the regenerated proof |
-| Five-judge verification | `KernelPipeline.verify_candidate` | Five `JudgeVerdict` objects per round |
-| Consecutive-pass protocol | `KernelPipeline.verify_candidate` | `K=2`; rejection resets the streak |
-| Repair loop | `KernelPipeline._repair` | Corrected candidate with a distinct content hash; maximum `T=15` rounds |
-| Audit sampling flag | `KernelPipeline._audit_selected` | Deterministic 10% post-hoc audit selection |
+| Surface rename | `SurfacePipeline.run_family` | `surface_<family>_variant.json` |
+| 1. Reference solution to proof structure | `KernelPipeline.extract_plan` | `01_proof_dag.json` |
+| 2. Content-free method plan | `KernelPipeline.extract_plan` | `02_method_plan.json` |
+| 3. Mutable-slot identification | `KernelPipeline.extract_plan` | `03_mutable_slots.json` |
+| 4. Proof regeneration | `KernelPipeline.generate_candidate` | `04_regenerated_proof.json` |
+| 5. Problem rendering | `KernelPipeline.generate_candidate` | `05_variant_question.json` |
+| Five-judge verification | `KernelPipeline.verify_candidate` | five call records and one iteration record per round |
+| Consecutive-pass protocol | `KernelPipeline.verify_candidate` | `K=2`; any rejection resets the streak |
+| Repair loop | `KernelPipeline._repair` | complete corrected question and solution; at most `T=15` rounds |
 
 ## Per-item artifacts
 
@@ -25,11 +30,6 @@ items/<item-id>/
   final.json
 ```
 
-Each JSON artifact includes a SHA-256 digest of its canonical payload. Judge
-calls remain separate rather than being collapsed into a vote count.
-
-## Claim boundary
-
-The verifier is a filtering protocol, not a formal proof of mathematical
-correctness. Accepted variants should additionally be inspected in the
-post-hoc human audit described in the accompanying paper.
+Prompt literals are byte-locked by `PROMPT_SHA256SUMS` and
+`tests/test_prompts.py`. The OpenAI adapter does not pass a temperature
+argument; `o3` therefore uses its supported default.
diff --git a/pyproject.toml b/pyproject.toml
index ce2afe9..bcfb1e3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "gap-reproduce"
-version = "0.1.0"
+version = "0.2.0"
 description = "Minimal reproduction package for the GAP mathematical transformation pipeline"
 requires-python = ">=3.11"
 dependencies = [
diff --git a/src/gap_pipeline/__init__.py b/src/gap_pipeline/__init__.py
index 2591afe..fe2e04c 100644
--- a/src/gap_pipeline/__init__.py
+++ b/src/gap_pipeline/__init__.py
@@ -3,10 +3,8 @@
 from .models import (
     CanonicalItem,
     KernelCandidate,
+    KernelPlan,
     KernelRunResult,
-    MethodPlan,
-    ProofDAG,
-    ReplacementSpec,
 )
 from .pipeline import KernelPipeline, PipelineConfig
 
@@ -14,11 +12,9 @@ __all__ = [
     "CanonicalItem",
     "KernelCandidate",
     "KernelPipeline",
+    "KernelPlan",
     "KernelRunResult",
-    "MethodPlan",
     "PipelineConfig",
-    "ProofDAG",
-    "ReplacementSpec",
 ]
 
-__version__ = "0.1.0"
+__version__ = "0.2.0"
diff --git a/src/gap_pipeline/e2e.py b/src/gap_pipeline/e2e.py
index 093068f..971da3e 100644
--- a/src/gap_pipeline/e2e.py
+++ b/src/gap_pipeline/e2e.py
@@ -1,4 +1,4 @@
-"""End-to-end helpers used by the reproducibility notebook."""
+"""End-to-end helpers used by the reproduction notebook."""
 
 from __future__ import annotations
 
@@ -14,7 +14,7 @@ from .offline import load_dataset
 from .pipeline import KernelPipeline, PipelineConfig
 from .release import export_release
 from .store import RunStore
-from .surface import SURFACE_FAMILIES, SurfacePipeline
+from .surface import SurfacePipeline
 
 
 def _ensure_fresh(path: Path) -> None:
@@ -22,113 +22,28 @@ def _ensure_fresh(path: Path) -> None:
         raise FileExistsError(f"refusing to reuse non-empty output directory {path}")
 
 
-def _accept_verdict() -> dict[str, Any]:
+def _review_accept() -> dict[str, str]:
     return {
         "verdict": "accept",
-        "step_by_step_check": "n1 and n2 instantiate the fixed method plan",
+        "step_by_step_check": "both method labels are instantiated",
         "blocking_issues": "",
         "patch_suggestion": "",
     }
 
 
-def _offline_fixture() -> dict[str, Any]:
-    plan = {
-        "steps": [
-            {
-                "node_id": "n1",
-                "method_label": "use nonnegativity of a square",
-                "input_roles": ["positive scalar"],
-                "output_role": "nonnegative expression",
-                "invariants": ["scalar is real"],
-            },
-            {
-                "node_id": "n2",
-                "method_label": "expand and divide by a positive quantity",
-                "input_roles": ["nonnegative expression", "positive scalar"],
-                "output_role": "target inequality",
-                "invariants": ["divisor is positive"],
-            },
-        ],
-        "terminal_node_id": "n2",
-    }
-    replacement = {
-        "target_node_id": "n1",
-        "original_object": "(a-1)^2",
-        "replacement_object": "(x-2)^2",
-        "guard_conditions": ["x>0"],
-        "guard_evidence": ["positive real x satisfies the division guard"],
-        "expected_downstream_changes": ["expand around 2", "divide by x"],
-        "rationale": "the square-nonnegativity plan is unchanged",
-    }
-    diffused = {
-        "node_instantiations": [
-            {
-                "node_id": "n1",
-                "method_label": plan["steps"][0]["method_label"],
-                "instantiated_claim": "(x-2)^2 >= 0",
-                "instantiated_derivation": "a square is nonnegative",
-                "dependencies": [],
-            },
-            {
-                "node_id": "n2",
-                "method_label": plan["steps"][1]["method_label"],
-                "instantiated_claim": "x+4/x >= 4",
-                "instantiated_derivation": "expand and divide by positive x",
-                "dependencies": ["n1"],
-            },
-        ],
-        "regenerated_proof": (
-            "Since (x-2)^2 >= 0, expansion and division by x>0 give "
-            "x+4/x >= 4."
-        ),
-        "terminal_answer": "x+4/x >= 4",
-    }
-    candidate = {
-        "problem": "Let x>0. Prove that x+4/x >= 4.",
-        "proof": diffused["regenerated_proof"],
-        "terminal_answer": diffused["terminal_answer"],
-        "node_instantiations": diffused["node_instantiations"],
-        "replacement": replacement,
-    }
+def _offline_record() -> dict[str, Any]:
     return {
-        "record": {
-            "index": "demo-A-1",
-            "question": r"Let \(a>0\). Prove that \(a+1/a\ge 2\).",
-            "solution": (
-                r"Since \((a-1)^2\ge0\), expand and divide by \(a>0\) "
-                r"to obtain \(a+1/a\ge2\)."
-            ),
-            "vars": ["a"],
-            "params": [],
-            "sci_consts": [],
-            "problem_type": "proof",
-            "variants": {},
-        },
-        "dag": {
-            "nodes": [
-                {
-                    "node_id": "n1",
-                    "claim": "(a-1)^2 >= 0",
-                    "derivation": "a square is nonnegative",
-                    "dependencies": [],
-                    "source_span": "Since (a-1)^2 >= 0",
-                    "mathematical_objects": ["a", "(a-1)^2"],
-                },
-                {
-                    "node_id": "n2",
-                    "claim": "a+1/a >= 2",
-                    "derivation": "expand and divide by positive a",
-                    "dependencies": ["n1"],
-                    "source_span": "divide by a>0",
-                    "mathematical_objects": ["a"],
-                },
-            ],
-            "terminal_node_id": "n2",
-        },
-        "plan": plan,
-        "replacement": replacement,
-        "diffused": diffused,
-        "candidate": candidate,
+        "index": "demo-A-1",
+        "question": r"Let \(a>0\). Prove that \(a+1/a\ge 2\).",
+        "solution": (
+            r"Since \((a-1)^2\ge0\), expand and divide by \(a>0\) "
+            r"to obtain \(a+1/a\ge2\)."
+        ),
+        "vars": ["a"],
+        "params": [],
+        "sci_consts": [],
+        "problem_type": "proof",
+        "variants": {},
     }
 
 
@@ -140,7 +55,7 @@ async def run_live_item(
     model: str = "o3",
     api_key: str | None = None,
 ) -> dict[str, Any]:
-    """Generate all five GAP variants for one Putnam item and export them."""
+    """Generate four surface variants and one verified kernel variant."""
 
     if not (api_key or os.getenv("OPENAI_API_KEY")):
         raise RuntimeError(
@@ -162,29 +77,23 @@ async def run_live_item(
     surface_store = RunStore(surface_root, item.item_id)
     surface_store.write_input(item)
     surface_store.write_config(
-        {
-            "protocol_name": "gap-surface-v1",
-            "proposer_model": model,
-        }
+        {"protocol_name": "gap-surface-original-prompts", "model": model}
     )
-    surface_pipeline = SurfacePipeline(
+    surfaces = await SurfacePipeline(
         OpenAIJsonClient(model, api_key=api_key),
         surface_store,
-    )
-    surfaces = await surface_pipeline.run_all(item)
+    ).run_all(item)
 
     config = PipelineConfig(proposer_model=model, judge_model=model)
-    kernel_pipeline = KernelPipeline(
+    kernel = await KernelPipeline(
         proposer=OpenAIJsonClient(model, api_key=api_key),
         judges=[OpenAIJsonClient(model, api_key=api_key) for _ in range(5)],
         store=RunStore(kernel_root, item.item_id),
         config=config,
-    )
-    kernel = await kernel_pipeline.run(item)
+    ).run(item)
     if kernel.status != "accepted":
         raise RuntimeError(
-            f"kernel candidate was rejected after {len(kernel.iterations)} rounds; "
-            f"inspect {kernel_root / 'items' / item.item_id}"
+            f"kernel candidate was rejected after {len(kernel.iterations)} rounds"
         )
 
     manifest = export_release(
@@ -211,7 +120,7 @@ async def run_live_item(
 
 
 async def run_offline_smoke(work_root: Path) -> dict[str, Any]:
-    """Exercise the same end-to-end code path with deterministic responses."""
+    """Exercise the end-to-end wiring with deterministic model responses."""
 
     _ensure_fresh(work_root)
     work_root.mkdir(parents=True, exist_ok=True)
@@ -221,8 +130,7 @@ async def run_offline_smoke(work_root: Path) -> dict[str, Any]:
     kernel_root = work_root / "kernel-runs"
     release_root = work_root / "release"
 
-    fixture = _offline_fixture()
-    record = fixture["record"]
+    record = _offline_record()
     item_id = str(record["index"])
     (source_root / f"{item_id}.json").write_text(
         json.dumps(record, ensure_ascii=False, indent=2) + "\n",
@@ -230,21 +138,24 @@ async def run_offline_smoke(work_root: Path) -> dict[str, Any]:
     )
     item = CanonicalItem.from_public_record(record)
 
+    names = {
+        "descriptive_long": "positivequantity",
+        "descriptive_long_confusing": "walnutvioletterrace",
+        "descriptive_long_misleading": "primefieldorder",
+        "garbled_string": "qzxwvtnphjgrksla",
+    }
     surface_responses = {
-        f"{item_id}.surface.{family}.a": {
-            "replacement": {
-                "descriptive_long": "positiveQuantity",
-                "descriptive_long_confusing": "walnutVioletTerrace",
-                "descriptive_long_misleading": "primeFieldOrder",
-                "garbled_string": "xcQ7h2ZfRw9v",
-            }[family]
+        f"{item_id}.surface.{family}": {
+            "map": {"a": name},
+            "question": record["question"].replace("a", name),
+            "solution": record["solution"].replace("a", name),
         }
-        for family in SURFACE_FAMILIES
+        for family, name in names.items()
     }
     surface_store = RunStore(surface_root, item_id)
     surface_store.write_input(item)
     surface_store.write_config(
-        {"protocol_name": "gap-surface-smoke", "proposer_model": "scripted"}
+        {"protocol_name": "gap-surface-smoke", "model": "scripted"}
     )
     surfaces = await SurfacePipeline(
         ScriptedClient(surface_responses),
@@ -253,21 +164,30 @@ async def run_offline_smoke(work_root: Path) -> dict[str, Any]:
 
     proposer = ScriptedClient(
         {
-            f"{item_id}.stage1": fixture["dag"],
-            f"{item_id}.stage2": fixture["plan"],
-            f"{item_id}.stage3": fixture["replacement"],
-            f"{item_id}.stage4": fixture["diffused"],
-            f"{item_id}.stage5": fixture["candidate"],
+            f"{item_id}.plan": {
+                "core_steps": [
+                    "use nonnegativity of a square",
+                    "expand and divide by a positive quantity",
+                ],
+                "mutable_slots": {
+                    "slot1": {
+                        "description": "the positive reference value",
+                        "original": "1",
+                    }
+                },
+            },
+            f"{item_id}.candidate": {
+                "question": "Let x>0. Prove that x+4/x >= 4.",
+                "solution": (
+                    "Since (x-2)^2 >= 0, expansion and division by x>0 "
+                    "give x+4/x >= 4."
+                ),
+            },
         }
     )
     judges = [
         ScriptedClient(
-            {
-                f"{item_id}.verify": [
-                    _accept_verdict(),
-                    _accept_verdict(),
-                ]
-            }
+            {f"{item_id}.verify": [_review_accept(), _review_accept()]}
         )
         for _ in range(5)
     ]
diff --git a/src/gap_pipeline/models.py b/src/gap_pipeline/models.py
index fe7aeb7..bf80c5b 100644
--- a/src/gap_pipeline/models.py
+++ b/src/gap_pipeline/models.py
@@ -1,4 +1,4 @@
-"""Validated schemas for every GAP pipeline boundary."""
+"""Validated schemas for the prompt-faithful GAP reproduction path."""
 
 from __future__ import annotations
 
@@ -8,7 +8,7 @@ from typing import Any, Literal
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 
-SCHEMA_VERSION = "gap-reference-v1"
+SCHEMA_VERSION = "gap-prompt-faithful-v1"
 
 
 class StrictModel(BaseModel):
@@ -49,177 +49,33 @@ class CanonicalItem(StrictModel):
         )
 
 
-class ProofNode(StrictModel):
-    node_id: str
-    claim: str
-    derivation: str
-    dependencies: list[str] = Field(default_factory=list)
-    source_span: str = ""
-    mathematical_objects: list[str] = Field(default_factory=list)
+class MutableSlot(StrictModel):
+    description: str
+    original: str
 
 
-class ProofDAG(StrictModel):
-    nodes: list[ProofNode]
-    terminal_node_id: str
+class KernelPlan(StrictModel):
+    core_steps: list[str] = Field(min_length=1, max_length=5)
+    mutable_slots: dict[str, MutableSlot]
 
     @model_validator(mode="after")
-    def validate_graph(self) -> "ProofDAG":
-        if not self.nodes:
-            raise ValueError("proof DAG must contain at least one node")
-        ids = [node.node_id for node in self.nodes]
-        if len(ids) != len(set(ids)):
-            raise ValueError("proof DAG node IDs must be unique")
-        known = set(ids)
-        if self.terminal_node_id not in known:
-            raise ValueError("terminal node is absent from proof DAG")
-        for node in self.nodes:
-            if node.node_id in node.dependencies:
-                raise ValueError(f"node {node.node_id} depends on itself")
-            unknown = set(node.dependencies) - known
-            if unknown:
-                raise ValueError(
-                    f"node {node.node_id} has unknown dependencies {sorted(unknown)}"
-                )
-
-        dependencies = {node.node_id: set(node.dependencies) for node in self.nodes}
-        remaining = set(known)
-        resolved: set[str] = set()
-        while remaining:
-            ready = {node_id for node_id in remaining if dependencies[node_id] <= resolved}
-            if not ready:
-                raise ValueError("proof graph contains a cycle")
-            resolved.update(ready)
-            remaining.difference_update(ready)
-
-        ancestors = {self.terminal_node_id}
-        frontier = [self.terminal_node_id]
-        while frontier:
-            node_id = frontier.pop()
-            for dependency in dependencies[node_id]:
-                if dependency not in ancestors:
-                    ancestors.add(dependency)
-                    frontier.append(dependency)
-        disconnected = known - ancestors
-        if disconnected:
-            raise ValueError(
-                "proof DAG contains nodes disconnected from the terminal "
-                f"conclusion: {sorted(disconnected)}"
-            )
+    def validate_content(self) -> "KernelPlan":
+        if any(not step.strip() for step in self.core_steps):
+            raise ValueError("core steps must be non-empty")
+        if not self.mutable_slots:
+            raise ValueError("at least one mutable slot is required")
         return self
 
-    def topological_nodes(self) -> list[ProofNode]:
-        by_id = {node.node_id: node for node in self.nodes}
-        remaining = set(by_id)
-        resolved: set[str] = set()
-        ordered: list[ProofNode] = []
-        while remaining:
-            ready = sorted(
-                node_id
-                for node_id in remaining
-                if set(by_id[node_id].dependencies) <= resolved
-            )
-            for node_id in ready:
-                ordered.append(by_id[node_id])
-                resolved.add(node_id)
-                remaining.remove(node_id)
-        return ordered
-
-    def leaf_node_ids(self) -> list[str]:
-        return sorted(node.node_id for node in self.nodes if not node.dependencies)
-
-
-class MethodStep(StrictModel):
-    node_id: str
-    method_label: str
-    input_roles: list[str] = Field(default_factory=list)
-    output_role: str
-    invariants: list[str] = Field(default_factory=list)
-
-
-class MethodPlan(StrictModel):
-    steps: list[MethodStep]
-    terminal_node_id: str
-
-    def validate_against(self, dag: ProofDAG) -> None:
-        dag_order = [node.node_id for node in dag.topological_nodes()]
-        plan_order = [step.node_id for step in self.steps]
-        if plan_order != dag_order:
-            raise ValueError(
-                "method plan must contain every DAG node in topological order; "
-                f"expected {dag_order}, received {plan_order}"
-            )
-        if self.terminal_node_id != dag.terminal_node_id:
-            raise ValueError("method-plan terminal does not match DAG terminal")
-        if any(not step.method_label.strip() for step in self.steps):
-            raise ValueError("method labels must be non-empty")
-
-
-class ReplacementSpec(StrictModel):
-    target_node_id: str
-    original_object: str
-    replacement_object: str
-    guard_conditions: list[str]
-    guard_evidence: list[str]
-    expected_downstream_changes: list[str]
-    rationale: str
-
-    def validate_against(self, dag: ProofDAG) -> None:
-        if self.target_node_id not in dag.leaf_node_ids():
-            raise ValueError(
-                f"replacement target {self.target_node_id} is not a DAG leaf"
-            )
-        if self.original_object.strip() == self.replacement_object.strip():
-            raise ValueError("replacement must genuinely change the leaf object")
-        if not self.guard_conditions or not self.guard_evidence:
-            raise ValueError("replacement needs explicit guards and guard evidence")
-
-
-class NodeInstantiation(StrictModel):
-    node_id: str
-    method_label: str
-    instantiated_claim: str
-    instantiated_derivation: str
-    dependencies: list[str] = Field(default_factory=list)
-
-
-class DiffusedProof(StrictModel):
-    node_instantiations: list[NodeInstantiation]
-    regenerated_proof: str
-    terminal_answer: str
-
-    def validate_against(self, dag: ProofDAG, plan: MethodPlan) -> None:
-        dag_order = [node.node_id for node in dag.topological_nodes()]
-        actual_order = [node.node_id for node in self.node_instantiations]
-        if actual_order != dag_order:
-            raise ValueError("diffused proof node order differs from proof DAG")
-        expected_labels = [step.method_label for step in plan.steps]
-        actual_labels = [node.method_label for node in self.node_instantiations]
-        if actual_labels != expected_labels:
-            raise ValueError("diffused proof changes the method-label sequence")
-        expected_dependencies = {
-            node.node_id: node.dependencies for node in dag.topological_nodes()
-        }
-        for node in self.node_instantiations:
-            if node.dependencies != expected_dependencies[node.node_id]:
-                raise ValueError(
-                    f"diffused proof changes dependencies for {node.node_id}"
-                )
-
 
 class KernelCandidate(StrictModel):
-    problem: str
-    proof: str
-    terminal_answer: str
-    node_instantiations: list[NodeInstantiation]
-    replacement: ReplacementSpec
+    question: str
+    solution: str
 
-    def validate_against(self, dag: ProofDAG, plan: MethodPlan) -> None:
-        DiffusedProof(
-            node_instantiations=self.node_instantiations,
-            regenerated_proof=self.proof,
-            terminal_answer=self.terminal_answer,
-        ).validate_against(dag, plan)
-        self.replacement.validate_against(dag)
+    @model_validator(mode="after")
+    def validate_content(self) -> "KernelCandidate":
+        if not self.question.strip() or not self.solution.strip():
+            raise ValueError("kernel question and solution must be non-empty")
+        return self
 
 
 class JudgeVerdict(StrictModel):
@@ -230,14 +86,10 @@ class JudgeVerdict(StrictModel):
 
     @model_validator(mode="after")
     def validate_verdict(self) -> "JudgeVerdict":
-        if self.verdict == "accept":
-            if self.blocking_issues.strip():
-                raise ValueError("accept verdict cannot contain blocking issues")
-        else:
-            if not self.blocking_issues.strip():
-                raise ValueError("reject verdict must identify a blocking issue")
-            if not self.patch_suggestion.strip():
-                raise ValueError("reject verdict must include a patch suggestion")
+        if self.verdict == "accept" and self.blocking_issues.strip():
+            raise ValueError("accept verdict cannot contain blocking issues")
+        if self.verdict == "reject" and not self.blocking_issues.strip():
+            raise ValueError("reject verdict must identify a blocking issue")
         return self
 
 
@@ -247,22 +99,22 @@ class IterationRecord(StrictModel):
     verdicts: list[JudgeVerdict]
     unanimous: bool
     pass_streak_after: int
-    repaired: bool = False
-    repaired_candidate_sha256: str | None = None
 
 
 class KernelRunResult(StrictModel):
     item_id: str
     status: Literal["accepted", "rejected"]
+    plan: KernelPlan
     accepted_candidate: KernelCandidate | None = None
     iterations: list[IterationRecord]
     rejection_reason: str = ""
+    accepted_candidate_sha256: str | None = None
     schema_version: str = SCHEMA_VERSION
 
     @model_validator(mode="after")
     def validate_terminal_state(self) -> "KernelRunResult":
         if self.status == "accepted" and self.accepted_candidate is None:
-            raise ValueError("accepted run must include its accepted candidate")
+            raise ValueError("accepted run must include its candidate")
         if self.status == "rejected" and not self.rejection_reason:
             raise ValueError("rejected run must state a reason")
         return self
diff --git a/src/gap_pipeline/pipeline.py b/src/gap_pipeline/pipeline.py
index 2c0deb4..307ba81 100644
--- a/src/gap_pipeline/pipeline.py
+++ b/src/gap_pipeline/pipeline.py
@@ -1,42 +1,30 @@
-"""Five-stage kernel generation and the J=5, K=2, T=15 verifier."""
+"""Prompt-faithful kernel generation and the J=5, K=2, T=15 loop."""
 
 from __future__ import annotations
 
 import asyncio
-import hashlib
-from pathlib import Path
-from typing import Any
 
 from pydantic import BaseModel, ConfigDict, model_validator
 
 from .clients import JsonLLM
 from .models import (
     CanonicalItem,
-    DiffusedProof,
     IterationRecord,
     JudgeVerdict,
     KernelCandidate,
+    KernelPlan,
     KernelRunResult,
-    MethodPlan,
     ModelCallRecord,
-    ProofDAG,
-    ReplacementSpec,
 )
 from .prompts import (
-    DAG_SYSTEM,
-    DIFFUSION_SYSTEM,
-    JUDGE_SYSTEM,
-    METHOD_SYSTEM,
-    QUESTION_SYSTEM,
-    REPAIR_SYSTEM,
-    REPLACEMENT_SYSTEM,
-    dag_user,
-    diffusion_user,
+    FIX_SYSTEM_PROMPT,
+    JUDGE_SYSTEM_PROMPT,
+    KERNEL_GENERATE_SYSTEM,
+    KERNEL_PLAN_SYSTEM,
+    fix_user,
     judge_user,
-    method_user,
-    question_user,
-    repair_user,
-    replacement_user,
+    kernel_generate_user,
+    kernel_plan_user,
 )
 from .store import RunStore, sha256_payload
 
@@ -50,22 +38,11 @@ class PipelineConfig(BaseModel):
     judge_count: int = 5
     streak_length: int = 2
     max_iterations: int = 15
-    human_audit_fraction: float = 0.10
-    audit_seed: int = 0
-    difficulty_instruction: str = (
-        "Preserve the proof plan and avoid trivializing the source problem. "
-        "Routine algebraic adaptation is allowed, but do not introduce a new "
-        "pivotal lemma or an independent proof strategy."
-    )
 
     @model_validator(mode="after")
-    def enforce_submitted_protocol(self) -> "PipelineConfig":
+    def enforce_protocol(self) -> "PipelineConfig":
         if (self.judge_count, self.streak_length, self.max_iterations) != (5, 2, 15):
-            raise ValueError(
-                "the GAP protocol is fixed at J=5, K=2, T=15"
-            )
-        if self.human_audit_fraction != 0.10:
-            raise ValueError("the GAP post-hoc audit fraction is 10%")
+            raise ValueError("the GAP protocol is fixed at J=5, K=2, T=15")
         return self
 
 
@@ -80,7 +57,7 @@ class KernelPipeline:
     ) -> None:
         if len(judges) != config.judge_count:
             raise ValueError(
-                f"expected {config.judge_count} judge clients, received {len(judges)}"
+                f"expected {config.judge_count} judges, received {len(judges)}"
             )
         self.proposer = proposer
         self.judges = judges
@@ -94,8 +71,7 @@ class KernelPipeline:
         request_id: str,
         system_prompt: str,
         user_prompt: str,
-        response_type: type[BaseModel],
-    ) -> BaseModel:
+    ) -> dict:
         response = await client.generate_json(
             system_prompt=system_prompt,
             user_prompt=user_prompt,
@@ -113,193 +89,125 @@ class KernelPipeline:
                 usage=response.usage,
             )
         )
-        return response_type.model_validate(response.data)
+        return response.data
 
-    async def construct_dag(self, item: CanonicalItem) -> ProofDAG:
-        request_id = f"{item.item_id}.stage1.dag"
-        dag = await self._call(
+    async def extract_plan(self, item: CanonicalItem) -> KernelPlan:
+        request_id = f"{item.item_id}.plan"
+        payload = await self._call(
             self.proposer,
             request_id=request_id,
-            system_prompt=DAG_SYSTEM,
-            user_prompt=dag_user(item),
-            response_type=ProofDAG,
+            system_prompt=KERNEL_PLAN_SYSTEM,
+            user_prompt=kernel_plan_user(item),
         )
-        assert isinstance(dag, ProofDAG)
-        self.store.write_stage("01_proof_dag", dag, request_id=request_id)
-        return dag
-
-    async def summarize_methods(self, item: CanonicalItem, dag: ProofDAG) -> MethodPlan:
-        request_id = f"{item.item_id}.stage2.methods"
-        plan = await self._call(
-            self.proposer,
+        plan = KernelPlan.model_validate(payload)
+        nodes = [
+            {
+                "node_id": f"n{index}",
+                "method_label": step,
+                "dependencies": [] if index == 1 else [f"n{index - 1}"],
+            }
+            for index, step in enumerate(plan.core_steps, start=1)
+        ]
+        self.store.write_stage(
+            "01_proof_dag",
+            {
+                "nodes": nodes,
+                "terminal_node_id": nodes[-1]["node_id"],
+                "construction": "ordered core_steps returned verbatim by Prompt-A",
+            },
             request_id=request_id,
-            system_prompt=METHOD_SYSTEM,
-            user_prompt=method_user(dag),
-            response_type=MethodPlan,
         )
-        assert isinstance(plan, MethodPlan)
-        plan.validate_against(dag)
-        self.store.write_stage("02_method_plan", plan, request_id=request_id)
-        return plan
-
-    async def generate_replacement(
-        self,
-        item: CanonicalItem,
-        dag: ProofDAG,
-        plan: MethodPlan,
-    ) -> ReplacementSpec:
-        request_id = f"{item.item_id}.stage3.replacement"
-        replacement = await self._call(
-            self.proposer,
+        self.store.write_stage(
+            "02_method_plan",
+            {"method_labels": plan.core_steps},
             request_id=request_id,
-            system_prompt=REPLACEMENT_SYSTEM,
-            user_prompt=replacement_user(
-                item,
-                dag,
-                plan,
-                difficulty_instruction=self.config.difficulty_instruction,
-            ),
-            response_type=ReplacementSpec,
         )
-        assert isinstance(replacement, ReplacementSpec)
-        replacement.validate_against(dag)
         self.store.write_stage(
-            "03_replacement",
-            replacement,
+            "03_mutable_slots",
+            {
+                key: value.model_dump(mode="json")
+                for key, value in plan.mutable_slots.items()
+            },
             request_id=request_id,
         )
-        return replacement
+        return plan
 
-    async def diffuse_dag(
+    async def generate_candidate(
         self,
         item: CanonicalItem,
-        dag: ProofDAG,
-        plan: MethodPlan,
-        replacement: ReplacementSpec,
-    ) -> DiffusedProof:
-        request_id = f"{item.item_id}.stage4.diffusion"
-        diffused = await self._call(
+        plan: KernelPlan,
+    ) -> KernelCandidate:
+        request_id = f"{item.item_id}.candidate"
+        payload = await self._call(
             self.proposer,
             request_id=request_id,
-            system_prompt=DIFFUSION_SYSTEM,
-            user_prompt=diffusion_user(item, dag, plan, replacement),
-            response_type=DiffusedProof,
+            system_prompt=KERNEL_GENERATE_SYSTEM,
+            user_prompt=kernel_generate_user(item, plan),
         )
-        assert isinstance(diffused, DiffusedProof)
-        diffused.validate_against(dag, plan)
+        candidate = KernelCandidate.model_validate(payload)
         self.store.write_stage(
-            "04_diffused_proof",
-            diffused,
+            "04_regenerated_proof",
+            {"solution": candidate.solution},
             request_id=request_id,
         )
-        return diffused
-
-    async def render_question(
-        self,
-        item: CanonicalItem,
-        dag: ProofDAG,
-        plan: MethodPlan,
-        replacement: ReplacementSpec,
-        diffused: DiffusedProof,
-    ) -> KernelCandidate:
-        request_id = f"{item.item_id}.stage5.question"
-        candidate = await self._call(
-            self.proposer,
-            request_id=request_id,
-            system_prompt=QUESTION_SYSTEM,
-            user_prompt=question_user(
-                item,
-                dag,
-                plan,
-                replacement,
-                diffused.model_dump(mode="json"),
-            ),
-            response_type=KernelCandidate,
-        )
-        assert isinstance(candidate, KernelCandidate)
-        candidate.validate_against(dag, plan)
         self.store.write_stage(
-            "05_draft_candidate",
-            candidate,
+            "05_variant_question",
+            {"question": candidate.question},
             request_id=request_id,
         )
         return candidate
 
-    async def build_candidate(
-        self,
-        item: CanonicalItem,
-    ) -> tuple[ProofDAG, MethodPlan, KernelCandidate]:
-        dag = await self.construct_dag(item)
-        plan = await self.summarize_methods(item, dag)
-        replacement = await self.generate_replacement(item, dag, plan)
-        diffused = await self.diffuse_dag(item, dag, plan, replacement)
-        candidate = await self.render_question(
-            item,
-            dag,
-            plan,
-            replacement,
-            diffused,
-        )
-        return dag, plan, candidate
-
     async def _judge_once(
         self,
         judge: JsonLLM,
         *,
         item: CanonicalItem,
-        plan: MethodPlan,
+        plan: KernelPlan,
         candidate: KernelCandidate,
         iteration: int,
         judge_id: int,
     ) -> JudgeVerdict:
         request_id = f"{item.item_id}.verify.t{iteration:02d}.j{judge_id}"
-        verdict = await self._call(
+        payload = await self._call(
             judge,
             request_id=request_id,
-            system_prompt=JUDGE_SYSTEM,
-            user_prompt=judge_user(
-                item,
-                plan,
-                candidate,
-                judge_id=judge_id,
-                iteration=iteration,
-            ),
-            response_type=JudgeVerdict,
+            system_prompt=JUDGE_SYSTEM_PROMPT,
+            user_prompt=judge_user(item, plan, candidate),
         )
-        assert isinstance(verdict, JudgeVerdict)
-        return verdict
+        return JudgeVerdict.model_validate(payload)
 
     async def _repair(
         self,
         *,
         item: CanonicalItem,
-        dag: ProofDAG,
-        plan: MethodPlan,
         candidate: KernelCandidate,
         verdicts: list[JudgeVerdict],
         iteration: int,
     ) -> KernelCandidate:
+        problem_issues: list[str] = []
+        solution_issues: list[str] = []
+        for verdict in verdicts:
+            if verdict.verdict == "reject":
+                problem_issues.append(verdict.blocking_issues)
+                solution_issues.append(
+                    verdict.patch_suggestion or verdict.blocking_issues
+                )
         request_id = f"{item.item_id}.repair.after_t{iteration:02d}"
-        issues = [
-            {
-                "judge_id": judge_id,
-                "blocking_issues": verdict.blocking_issues,
-                "patch_suggestion": verdict.patch_suggestion,
-            }
-            for judge_id, verdict in enumerate(verdicts, start=1)
-            if verdict.verdict == "reject"
-        ]
-        repaired = await self._call(
+        payload = await self._call(
             self.proposer,
             request_id=request_id,
-            system_prompt=REPAIR_SYSTEM,
-            user_prompt=repair_user(item, dag, plan, candidate, issues),
-            response_type=KernelCandidate,
+            system_prompt=FIX_SYSTEM_PROMPT,
+            user_prompt=fix_user(
+                item,
+                candidate,
+                problem_issues="; ".join(problem_issues),
+                solution_issues="; ".join(solution_issues),
+            ),
+        )
+        repaired = KernelCandidate(
+            question=str(payload["corrected_question"]),
+            solution=str(payload["corrected_solution"]),
         )
-        assert isinstance(repaired, KernelCandidate)
-        repaired.validate_against(dag, plan)
-        if sha256_payload(repaired) == sha256_payload(candidate):
-            raise ValueError("repair call returned a byte-equivalent candidate")
         self.store.write_stage(
             f"repair_after_{iteration:02d}",
             repaired,
@@ -307,25 +215,17 @@ class KernelPipeline:
         )
         return repaired
 
-    def _audit_selected(self, item_id: str) -> bool:
-        digest = hashlib.sha256(
-            f"{self.config.audit_seed}:{item_id}".encode("utf-8")
-        ).digest()
-        draw = int.from_bytes(digest[:8], "big") / float(2**64)
-        return draw < self.config.human_audit_fraction
-
     async def verify_candidate(
         self,
         *,
         item: CanonicalItem,
-        dag: ProofDAG,
-        plan: MethodPlan,
+        plan: KernelPlan,
         candidate: KernelCandidate,
     ) -> KernelRunResult:
+        current = candidate
         iterations: list[IterationRecord] = []
         pass_streak = 0
-        current = candidate
-        streak_candidate_sha: str | None = None
+        streak_sha: str | None = None
 
         for iteration in range(1, self.config.max_iterations + 1):
             current_sha = sha256_payload(current)
@@ -345,115 +245,62 @@ class KernelPipeline:
                 )
             )
             unanimous = all(verdict.verdict == "accept" for verdict in verdicts)
-
             if unanimous:
-                if streak_candidate_sha not in {None, current_sha}:
+                if streak_sha not in {None, current_sha}:
                     raise AssertionError("pass streak crossed candidate versions")
-                streak_candidate_sha = current_sha
+                streak_sha = current_sha
                 pass_streak += 1
-                record = IterationRecord(
-                    iteration=iteration,
-                    candidate_sha256=current_sha,
-                    verdicts=verdicts,
-                    unanimous=True,
-                    pass_streak_after=pass_streak,
-                )
-                iterations.append(record)
-                self.store.write_iteration(iteration, record)
-                if pass_streak >= self.config.streak_length:
-                    result = KernelRunResult(
-                        item_id=item.item_id,
-                        status="accepted",
-                        accepted_candidate=current,
-                        iterations=iterations,
-                    )
-                    self.store.write_final(
-                        {
-                            **result.model_dump(mode="json"),
-                            "accepted_candidate_sha256": current_sha,
-                            "human_audit_selected": self._audit_selected(item.item_id),
-                        }
-                    )
-                    return result
-                continue
+            else:
+                pass_streak = 0
+                streak_sha = None
 
-            pass_streak = 0
-            streak_candidate_sha = None
-            if iteration == self.config.max_iterations:
-                record = IterationRecord(
-                    iteration=iteration,
-                    candidate_sha256=current_sha,
-                    verdicts=verdicts,
-                    unanimous=False,
-                    pass_streak_after=0,
-                )
-                iterations.append(record)
-                self.store.write_iteration(iteration, record)
-                break
-
-            repaired = await self._repair(
-                item=item,
-                dag=dag,
-                plan=plan,
-                candidate=current,
-                verdicts=verdicts,
-                iteration=iteration,
-            )
-            repaired_sha = sha256_payload(repaired)
             record = IterationRecord(
                 iteration=iteration,
                 candidate_sha256=current_sha,
                 verdicts=verdicts,
-                unanimous=False,
-                pass_streak_after=0,
-                repaired=True,
-                repaired_candidate_sha256=repaired_sha,
+                unanimous=unanimous,
+                pass_streak_after=pass_streak,
             )
             iterations.append(record)
             self.store.write_iteration(iteration, record)
-            current = repaired
+
+            if pass_streak == self.config.streak_length:
+                result = KernelRunResult(
+                    item_id=item.item_id,
+                    status="accepted",
+                    plan=plan,
+                    accepted_candidate=current,
+                    iterations=iterations,
+                    accepted_candidate_sha256=current_sha,
+                )
+                self.store.write_final(result)
+                return result
+
+            if not unanimous and iteration < self.config.max_iterations:
+                current = await self._repair(
+                    item=item,
+                    candidate=current,
+                    verdicts=verdicts,
+                    iteration=iteration,
+                )
 
         result = KernelRunResult(
             item_id=item.item_id,
             status="rejected",
+            plan=plan,
             iterations=iterations,
-            rejection_reason=(
-                f"no two consecutive unanimous passes within "
-                f"{self.config.max_iterations} iterations"
-            ),
-        )
-        self.store.write_final(
-            {
-                **result.model_dump(mode="json"),
-                "human_audit_selected": False,
-            }
+            rejection_reason="no two consecutive unanimous rounds within T=15",
         )
+        self.store.write_final(result)
         return result
 
     async def run(self, item: CanonicalItem) -> KernelRunResult:
         self.store.write_input(item)
         self.store.write_config(self.config)
-        dag, plan, candidate = await self.build_candidate(item)
+        plan = await self.extract_plan(item)
+        candidate = await self.generate_candidate(item, plan)
         return await self.verify_candidate(
             item=item,
-            dag=dag,
             plan=plan,
             candidate=candidate,
         )
-
-
-def make_pipeline(
-    *,
-    proposer: JsonLLM,
-    judge_factory: Any,
-    run_dir: Path,
-    item_id: str,
-    config: PipelineConfig,
-) -> KernelPipeline:
-    judges = [judge_factory(judge_id) for judge_id in range(1, 6)]
-    return KernelPipeline(
-        proposer=proposer,
-        judges=judges,
-        store=RunStore(run_dir, item_id),
-        config=config,
-    )
diff --git a/src/gap_pipeline/prompts.py b/src/gap_pipeline/prompts.py
index 667a47a..2ba87cf 100644
--- a/src/gap_pipeline/prompts.py
+++ b/src/gap_pipeline/prompts.py
@@ -1,338 +1,273 @@
-"""Prompts and strict JSON contracts for the GAP pipeline."""
+"""Verbatim prompts recovered from the original GAP/Putnam source.
+
+Do not edit prompt literals in this file. ``tests/test_prompts.py`` pins their
+SHA-256 digests against the recovered source files.
+"""
 
 from __future__ import annotations
 
 import json
 
-from .models import (
-    CanonicalItem,
-    KernelCandidate,
-    MethodPlan,
-    ProofDAG,
-    ReplacementSpec,
-)
-
-
-DAG_SYSTEM = """You are a mathematical proof-structure parser.
-Convert a supplied reference solution into a directed acyclic graph. Each node
-must be one locally checkable intermediate claim. Edges point from prerequisites
-to claims that use them. Preserve every essential proof step and do not invent
-new mathematics. Return JSON only."""
-
-
-def dag_user(item: CanonicalItem) -> str:
-    return f"""ITEM ID: {item.item_id}
-
-PROBLEM:
-{item.problem}
-
-REFERENCE SOLUTION:
-{item.solution}
-
-Return:
-{{
-  "nodes": [
-    {{
-      "node_id": "n1",
-      "claim": "concrete intermediate mathematical claim",
-      "derivation": "why this claim follows",
-      "dependencies": [],
-      "source_span": "corresponding reference-solution text",
-      "mathematical_objects": ["objects used at this node"]
-    }}
-  ],
-  "terminal_node_id": "node containing the requested conclusion"
-}}
-
-Requirements:
-- every dependency must refer to an earlier logical prerequisite;
-- the graph must be acyclic and connected to the terminal conclusion;
-- leaf nodes must expose the initial mathematical objects that could be
-  replaced to produce a genuinely new setting;
-- retain all essential cases, guards, and endpoint checks."""
-
-
-METHOD_SYSTEM = """You abstract concrete proof steps into a reusable proof plan.
-For every DAG node, produce a content-light method label that says what operation
-or lemma is used without copying the node's particular constants or object names.
-Keep exactly the DAG's topological order and node IDs. Return JSON only."""
-
-
-def method_user(dag: ProofDAG) -> str:
-    return f"""PROOF DAG:
-{dag.model_dump_json(indent=2)}
-
-Return:
-{{
-  "steps": [
-    {{
-      "node_id": "n1",
-      "method_label": "content-free operation or lemma",
-      "input_roles": ["abstract role consumed by this step"],
-      "output_role": "abstract role produced by this step",
-      "invariants": ["conditions that must remain true"]
-    }}
-  ],
-  "terminal_node_id": "{dag.terminal_node_id}"
-}}
-
-Do not merge, omit, reorder, or add nodes."""
-
-
-REPLACEMENT_SYSTEM = """You design one guarded replacement at a leaf of a proof
-DAG. The replacement must create a genuinely different mathematical setting
-while allowing the exact method-label plan to remain executable. Prefer a
-replacement that does not trivialize the problem. State every guard and cite
-concrete evidence from the source inequalities or assumptions. Return JSON only."""
-
-
-def replacement_user(
-    item: CanonicalItem,
-    dag: ProofDAG,
-    plan: MethodPlan,
-    *,
-    difficulty_instruction: str,
-) -> str:
-    return f"""ORIGINAL PROBLEM:
-{item.problem}
-
-ORIGINAL SOLUTION:
-{item.solution}
-
-PROOF DAG:
-{dag.model_dump_json(indent=2)}
-
-METHOD PLAN:
-{plan.model_dump_json(indent=2)}
-
-DIFFICULTY INSTRUCTION:
-{difficulty_instruction}
-
-Return:
-{{
-  "target_node_id": "a leaf node ID",
-  "original_object": "leaf object being replaced",
-  "replacement_object": "new object or setting",
-  "guard_conditions": ["condition required for every downstream step"],
-  "guard_evidence": ["source-derived evidence that the guard is satisfiable"],
-  "expected_downstream_changes": ["concrete changes expected during diffusion"],
-  "rationale": "why the same plan remains valid and the task is not trivialized"
-}}"""
-
-
-DIFFUSION_SYSTEM = """You re-instantiate a fixed method plan after one guarded
-leaf replacement. Propagate the replacement through every dependent DAG node.
-You must keep exactly the original node order, dependency structure, and method
-labels. All algebra, cases, domains, and endpoint checks must be valid in the
-new setting. Return JSON only."""
-
+from .models import CanonicalItem, KernelCandidate, KernelPlan
 
-def diffusion_user(
-    item: CanonicalItem,
-    dag: ProofDAG,
-    plan: MethodPlan,
-    replacement: ReplacementSpec,
-) -> str:
-    return f"""ORIGINAL PROBLEM:
-{item.problem}
-
-ORIGINAL REFERENCE SOLUTION:
-{item.solution}
-
-PROOF DAG:
-{dag.model_dump_json(indent=2)}
 
-FIXED METHOD PLAN:
-{plan.model_dump_json(indent=2)}
+# Source: PutnamVariants@c3bed737370df2dbf73afd66bf6e86d4ece82d68
+#         scripts/o3_kernel_variant.py
+KERNEL_PLAN_SYSTEM = "You are an IMO medalist & pedagogue."
+KERNEL_PLAN_PROMPT = """
+You are a competition-math expert.
 
-GUARDED REPLACEMENT:
-{replacement.model_dump_json(indent=2)}
+(1) Read the Putnam problem and its official solution below.
+(2) List the MINIMAL chain of lemmas / techniques essential to the solution.
+(3) Identify every numerical or structural element that could be changed
+    *without* altering that chain of reasoning.  Denote them as MUTABLE_SLOTS.
 
-Return:
+Return **one JSON object only**:
 {{
-  "node_instantiations": [
-    {{
-      "node_id": "same node ID",
-      "method_label": "exact corresponding method label",
-      "instantiated_claim": "new concrete claim",
-      "instantiated_derivation": "valid derivation in the new setting",
-      "dependencies": ["same dependency IDs"]
-    }}
-  ],
-  "regenerated_proof": "complete rigorous proof in the new setting",
-  "terminal_answer": "terminal conclusion established by that proof"
+  "core_steps": ["..."],                // 1–5 concise phrases
+  "mutable_slots": {{
+     "slot1": {{"description": "...", "original": "..."}},
+     "slot2": {{"description": "...", "original": "..."}}
+  }}
 }}
 
-The sequence and labels must match exactly. Do not write a problem statement
-yet; work forward from the replacement to a correct terminal answer."""
-
-
-QUESTION_SYSTEM = """You reverse-engineer a self-contained competition
-mathematics problem from a fully regenerated proof and terminal answer. State
-exactly the assumptions used by the proof, ask exactly for its terminal
-conclusion, and do not expose the solution strategy. Return JSON only."""
+PROBLEM:
+<<<{question}>>>
 
+SOLUTION:
+<<<{solution}>>>
+"""
 
-def question_user(
-    item: CanonicalItem,
-    dag: ProofDAG,
-    plan: MethodPlan,
-    replacement: ReplacementSpec,
-    diffused_payload: dict,
-) -> str:
-    return f"""SOURCE PROBLEM (style reference only):
-{item.problem}
+KERNEL_GENERATE_SYSTEM = "You are a creative yet rigorous math professor."
+KERNEL_GENERATE_PROMPT = """
+We previously extracted:
 
-FIXED METHOD PLAN:
-{plan.model_dump_json(indent=2)}
+CORE_STEPS = {core}
+MUTABLE_SLOTS = {slots}
 
-REPLACEMENT:
-{replacement.model_dump_json(indent=2)}
+Here is the ORIGINAL problem for reference:
+<<<{orig_q}>>>
 
-REGENERATED PROOF:
-{json.dumps(diffused_payload, ensure_ascii=False, indent=2)}
+And its OFFICIAL solution:
+<<<{orig_s}>>>
 
-Return:
+Create ONE *new* Putnam-level problem that
+  • still requires exactly the chain CORE_STEPS to solve,
+  • alters *every* MUTABLE_SLOT in a significant way.
+Return JSON only:
 {{
-  "problem": "self-contained candidate problem",
-  "proof": "the complete regenerated proof",
-  "terminal_answer": "answer requested by the candidate problem",
-  "node_instantiations": {json.dumps(diffused_payload.get("node_instantiations", []), ensure_ascii=False)},
-  "replacement": {replacement.model_dump_json()}
+ "question": "...",   // full statement (LaTeX-friendly)
+ "solution": "..."    // complete proof with new data
 }}
-
-The problem must be genuinely different from the source and must neither omit
-an assumption used by the proof nor add an assumption that trivializes it."""
+"""
 
 
-# Four-field verifier JSON contract used by the GAP pipeline.
-JUDGE_SYSTEM = """You are a verification judge for a kernel-variant generation
-pipeline. You must decide whether a CANDIDATE variant problem and its CANDIDATE
-proof are mathematically equivalent to the ORIGINAL problem under the given
-METHOD-LABEL sequence (the abstract proof plan).
+# Source: paper Appendix F.3, Listing 4.
+JUDGE_SYSTEM_PROMPT = """You are a verification judge for a kernel-variant generation pipeline. You must decide whether a CANDIDATE variant problem and its CANDIDATE proof are mathematically equivalent to the ORIGINAL problem under the given METHOD-LABEL sequence (the abstract proof plan).
 
 Your job is verification, not solving. You receive:
 - the ORIGINAL problem statement and reference solution,
-- the abstract METHOD-LABEL sequence,
+- the abstract METHOD-LABEL sequence (a list of content-free steps),
 - the SLOT replacement that was applied,
-- the CANDIDATE variant statement and regenerated proof.
-
-Check step by step that:
-1. Every candidate step instantiates the corresponding method label.
-2. The candidate proof is valid, with no unjustified leap.
-3. The candidate problem is well posed and its requested terminal answer
-   matches the regenerated proof.
-4. The candidate is genuinely different but uses the same plan.
-
-Return one JSON object only."""
+- the CANDIDATE variant statement and the CANDIDATE regenerated proof.
 
+You must check, step by step, that:
+1. Every CANDIDATE step instantiates the corresponding METHOD label with the new operands.
+2. The CANDIDATE proof is mathematically valid: each step follows from the previous one, with no unjustified leap.
+3. The CANDIDATE problem statement is well-posed and has a unique terminal answer matching the regenerated proof.
+4. The CANDIDATE variant is genuinely different from the ORIGINAL (the slot replacement actually changed the instance) but uses the same plan.
 
-def judge_user(
-    item: CanonicalItem,
-    plan: MethodPlan,
-    candidate: KernelCandidate,
-    *,
-    judge_id: int,
-    iteration: int,
-) -> str:
-    return f"""JUDGE REPLICATE: {judge_id}
-ITERATION: {iteration}
+Output a single JSON object with exactly these fields."""
 
-ORIGINAL PROBLEM:
-{item.problem}
+JUDGE_USER_TEMPLATE = """ORIGINAL PROBLEM:
+{original_problem}
 
 ORIGINAL REFERENCE SOLUTION:
-{item.solution}
+{original_solution}
 
-METHOD-LABEL SEQUENCE:
-{plan.model_dump_json(indent=2)}
+METHOD-LABEL SEQUENCE (abstract plan):
+{method_labels}
 
 SLOT REPLACEMENT:
-{candidate.replacement.model_dump_json(indent=2)}
+{slot_replacement}
 
 CANDIDATE VARIANT PROBLEM:
-{candidate.problem}
+{candidate_problem}
 
 CANDIDATE REGENERATED PROOF:
-{candidate.proof}
-
-CANDIDATE NODE INSTANTIATIONS:
-{json.dumps([node.model_dump() for node in candidate.node_instantiations], ensure_ascii=False, indent=2)}
+{candidate_proof}
 
 Return:
-{{
-  "verdict": "accept or reject",
-  "step_by_step_check": "for every METHOD label, state whether the candidate step instantiates it correctly",
-  "blocking_issues": "logical gap, computation error, ill-posed statement, or plan deviation; empty if accepted",
-  "patch_suggestion": "minimal correction if rejected; empty if accepted"
-}}
-
-The step-by-step check must explicitly cover every method-plan node."""
-
-
-REPAIR_SYSTEM = """You repair a rejected kernel candidate. Apply only the
-minimal changes needed to address all blocking issues while preserving the
-guarded replacement, DAG node order, dependencies, and exact method-label
-sequence. Return the complete corrected candidate as JSON only."""
+{{"verdict": "accept" or "reject",
+  "step_by_step_check": "for each METHOD label, state whether the CANDIDATE step instantiates it correctly",
+  "blocking_issues": "list any logical gap, computation error, ill-posed statement, or plan deviation",
+  "patch_suggestion": "if reject, propose a minimal patch (a corrected proof step or a corrected slot value); leave empty if accept"}}"""
+
+FIX_SYSTEM_PROMPT = """You are a mathematical expert tasked with fixing kernel variant problems.
+Based on the review feedback, correct the identified issues while maintaining the problem's essence.
+
+Guidelines:
+- Fix mathematical errors while preserving the problem's structure
+- Ensure the corrected version is well-posed and solvable
+- Keep solutions detailed and pedagogically clear
+- Maintain similar difficulty level to the original problem
+
+Provide COMPLETE corrected versions, not just patches."""
+
+FIX_USER_TEMPLATE = """Based on the review feedback, please fix this kernel variant:
+
+CURRENT PROBLEM:
+{kv_question}
+
+CURRENT SOLUTION:
+{kv_solution}
+
+REVIEW FEEDBACK:
+Problem Issues: {problem_issues}
+Solution Issues: {solution_issues}
+
+ORIGINAL PROBLEM (for reference):
+{orig_question}
+
+ORIGINAL SOLUTION (for reference):
+{orig_solution}
+
+Please provide corrected versions. Return JSON with:
+{{"corrected_question": "complete corrected problem statement",
+  "corrected_solution": "complete corrected solution",\x20
+  "changes_made": "summary of key changes made"}}"""
+
+
+# Source: PutnamVariants@c3bed737370df2dbf73afd66bf6e86d4ece82d68
+#         scripts/o3_rename_vars.py
+SURFACE_SYSTEM_BASE = "You are a meticulous LaTeX editor."
+SURFACE_TASK_COMMON = """
+Given:
+ • A Putnam problem statement and its official solution (LaTeX-like);
+ • Two symbol lists: vars (unknowns) and params (given constants).
+
+Rename every symbol in *vars* and *params* with a unique English identifier:
+  – all-lowercase letters, ≥8 chars, no underscore/space;
+  – same original symbol → same new name everywhere;
+  – different symbols → different new names;
+  – NEVER touch sci_consts (\\pi,e,i,…) or numeric constants;
+  – do NOT alter any other text or LaTeX markup.
+"""
+SURFACE_TASK_DESCRIPTIVE = """
+Each new identifier **should describe the symbol's mathematical role**.
+"""
+SURFACE_TASK_CONFUSING = """
+Each new identifier **should *not* match the symbol's role**; choose plausible
+but misleading nouns so the name sounds related but not matching the true meaning, such as replace Area with Radius.
+"""
+SURFACE_TASK_MISLEADING = """
+Each new identifier **should describe the *opposite* concept**. Pick names that
+directly contradict the symbol's actual meaning, e.g. rename a parallel vector
+to orthogonalvector.
+"""
+SURFACE_TASK_GARBLED = """
+Each new identifier **should look like random gibberish**: at least eight
+lowercase letters with no apparent meaning such as qzxwvtnp or hjgrksla.
+"""
+SURFACE_RETURN_SPEC = """
+Return exactly **one JSON object** and nothing else:
+{"map":{"old":"new",...},"question":"...","solution":"..."}
+"""
+SURFACE_USER_TEMPLATE = """Problem:
+<<<
+{question}
+>>>
+
+Solution:
+<<<
+{solution}
+>>>
+
+vars = {vars}
+params = {params}
+"""
+
+SURFACE_TASKS = {
+    "descriptive_long": SURFACE_TASK_DESCRIPTIVE,
+    "descriptive_long_confusing": SURFACE_TASK_CONFUSING,
+    "descriptive_long_misleading": SURFACE_TASK_MISLEADING,
+    "garbled_string": SURFACE_TASK_GARBLED,
+}
+
+
+def kernel_plan_user(item: CanonicalItem) -> str:
+    return KERNEL_PLAN_PROMPT.format(
+        question=item.problem,
+        solution=item.solution,
+    )
+
+
+def kernel_generate_user(item: CanonicalItem, plan: KernelPlan) -> str:
+    return KERNEL_GENERATE_PROMPT.format(
+        core=json.dumps(plan.core_steps, ensure_ascii=False),
+        slots=json.dumps(
+            {
+                key: value.model_dump(mode="json")
+                for key, value in plan.mutable_slots.items()
+            },
+            ensure_ascii=False,
+        ),
+        orig_q=item.problem,
+        orig_s=item.solution,
+    )
 
 
-def repair_user(
+def judge_user(
     item: CanonicalItem,
-    dag: ProofDAG,
-    plan: MethodPlan,
+    plan: KernelPlan,
     candidate: KernelCandidate,
-    issues: list[dict],
 ) -> str:
-    return f"""ORIGINAL PROBLEM:
-{item.problem}
-
-PROOF DAG:
-{dag.model_dump_json(indent=2)}
-
-FIXED METHOD PLAN:
-{plan.model_dump_json(indent=2)}
-
-CURRENT CANDIDATE:
-{candidate.model_dump_json(indent=2)}
-
-JUDGE FEEDBACK:
-{json.dumps(issues, ensure_ascii=False, indent=2)}
-
-Return the complete corrected object with fields:
-problem, proof, terminal_answer, node_instantiations, replacement.
-Do not change the replacement unless a judge proves it violates a guard; if it
-must change, retain the same target leaf and explain the corrected guards in
-the replacement rationale."""
-
-
-SURFACE_NAME_SYSTEM = """You propose identifier replacements for a mathematical
-problem. A replacement must be a single ASCII identifier beginning with a
-letter, must not collide with supplied identifiers, and must fit the requested
-family. Return JSON only."""
+    return JUDGE_USER_TEMPLATE.format(
+        original_problem=item.problem,
+        original_solution=item.solution,
+        method_labels=json.dumps(plan.core_steps, ensure_ascii=False),
+        slot_replacement=json.dumps(
+            {
+                key: value.model_dump(mode="json")
+                for key, value in plan.mutable_slots.items()
+            },
+            ensure_ascii=False,
+        ),
+        candidate_problem=candidate.question,
+        candidate_proof=candidate.solution,
+    )
 
 
-def surface_name_user(
+def fix_user(
+    item: CanonicalItem,
+    candidate: KernelCandidate,
     *,
-    symbol: str,
-    role: str,
-    family: str,
-    context: str,
-    forbidden: list[str],
+    problem_issues: str,
+    solution_issues: str,
 ) -> str:
-    descriptions = {
-        "descriptive_long": "a semantically helpful descriptive identifier",
-        "descriptive_long_confusing": "2-5 unrelated common words concatenated",
-        "descriptive_long_misleading": (
-            "mathematical jargon suggesting a different role"
-        ),
-        "garbled_string": "a 4-16 character semantically meaningless string",
-    }
-    return f"""SYMBOL: {symbol}
-ROLE: {role}
-FAMILY: {family} ({descriptions[family]})
-CONTEXT:
-{context}
-FORBIDDEN IDENTIFIERS:
-{json.dumps(forbidden)}
-
-Return {{"replacement": "one valid identifier", "rationale": "brief reason"}}."""
+    return FIX_USER_TEMPLATE.format(
+        kv_question=candidate.question,
+        kv_solution=candidate.solution,
+        problem_issues=problem_issues,
+        solution_issues=solution_issues,
+        orig_question=item.problem,
+        orig_solution=item.solution,
+    )
+
+
+def surface_system(family: str) -> str:
+    return (
+        SURFACE_SYSTEM_BASE
+        + SURFACE_TASK_COMMON
+        + SURFACE_TASKS[family]
+        + SURFACE_RETURN_SPEC
+    )
+
+
+def surface_user(item: CanonicalItem) -> str:
+    return SURFACE_USER_TEMPLATE.format(
+        question=item.problem,
+        solution=item.solution,
+        vars=item.variables,
+        params=item.parameters,
+    )
diff --git a/src/gap_pipeline/release.py b/src/gap_pipeline/release.py
index f86ce64..9afa117 100644
--- a/src/gap_pipeline/release.py
+++ b/src/gap_pipeline/release.py
@@ -90,8 +90,8 @@ def export_release(
 
         candidate = kernel_payload["accepted_candidate"]
         variants["kernel_variant"] = {
-            "question": candidate["problem"],
-            "solution": candidate["proof"],
+            "question": candidate["question"],
+            "solution": candidate["solution"],
         }
         output_record = copy.deepcopy(record)
         output_record["variants"] = variants
diff --git a/src/gap_pipeline/surface.py b/src/gap_pipeline/surface.py
index 7de4e90..914d847 100644
--- a/src/gap_pipeline/surface.py
+++ b/src/gap_pipeline/surface.py
@@ -1,16 +1,14 @@
-"""Surface-renaming generation, application, and deterministic validation."""
+"""Surface-renaming generation using the verbatim original prompts."""
 
 from __future__ import annotations
 
-import asyncio
-import hashlib
 import re
 from dataclasses import dataclass
 from typing import Literal
 
 from .clients import JsonLLM
 from .models import CanonicalItem, ModelCallRecord
-from .prompts import SURFACE_NAME_SYSTEM, surface_name_user
+from .prompts import surface_system, surface_user
 from .store import RunStore
 
 
@@ -28,99 +26,49 @@ SURFACE_FAMILIES: tuple[SurfaceFamily, ...] = (
     "garbled_string",
 )
 
-IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9]{2,63}$")
-MATH_SPAN_RE = re.compile(
-    r"(?P<dollar>\${1,2}.*?\${1,2})"
-    r"|(?P<paren>\\\(.*?\\\))"
-    r"|(?P<bracket>\\\[.*?\\\])"
-    r"|(?P<env>\\begin\{(?P<envname>[A-Za-z*]+)\}.*?\\end\{(?P=envname)\})",
-    flags=re.DOTALL,
-)
+IDENTIFIER_RE = re.compile(r"^[a-z]{8,}$")
 
 
 @dataclass(frozen=True)
 class SurfaceVariant:
     family: SurfaceFamily
     rename_map: dict[str, str]
-    problem: str
+    question: str
     solution: str
 
     def as_release_payload(self) -> dict[str, object]:
         return {
             "map": dict(self.rename_map),
-            "question": self.problem,
+            "question": self.question,
             "solution": self.solution,
         }
 
 
-def validate_rename_map(
-    rename_map: dict[str, str],
+def validate_surface_variant(
+    item: CanonicalItem,
     *,
-    existing_identifiers: list[str],
-    scientific_constants: list[str],
+    rename_map: dict[str, str],
+    question: str,
+    solution: str,
 ) -> None:
-    if not rename_map:
-        raise ValueError("surface rename map cannot be empty")
+    expected = set(item.variables + item.parameters)
+    if set(rename_map) != expected:
+        raise ValueError(
+            "rename map must contain every var and param exactly once; "
+            f"expected {sorted(expected)}, received {sorted(rename_map)}"
+        )
     if len(rename_map.values()) != len(set(rename_map.values())):
         raise ValueError("surface replacements must be one-to-one")
-    originals = set(existing_identifiers) | set(scientific_constants)
-    for old, new in rename_map.items():
-        if old not in existing_identifiers:
-            raise ValueError(f"rename source {old!r} is not in the symbol inventory")
-        if new in originals:
-            raise ValueError(f"replacement {new!r} collides with an existing identifier")
-        if not IDENTIFIER_RE.fullmatch(new):
-            raise ValueError(f"invalid replacement identifier {new!r}")
-        if new.startswith("\\"):
-            raise ValueError("replacement may not be a LaTeX command")
-
-
-def _replace_in_span(span: str, rename_map: dict[str, str]) -> str:
-    result = span
-    for old in sorted(rename_map, key=lambda value: (-len(value), value)):
-        # ASCII identifier boundaries avoid r -> radius changing \sqrt or prose.
-        pattern = re.compile(
-            rf"(?<![A-Za-z0-9]){re.escape(old)}(?![A-Za-z0-9])"
-        )
-        result = pattern.sub(lambda _: rename_map[old], result)
-    return result
-
-
-def apply_rename_map(text: str, rename_map: dict[str, str]) -> str:
-    """Rename identifiers only inside explicit LaTeX math spans.
-
-    This deliberately avoids the old prototype's ``a`` -> identifier mutation
-    in ordinary English prose. PutnamGAP source records use explicit math
-    delimiters for mathematical identifiers.
-    """
-
-    output: list[str] = []
-    cursor = 0
-    for match in MATH_SPAN_RE.finditer(text):
-        output.append(text[cursor : match.start()])
-        output.append(_replace_in_span(match.group(0), rename_map))
-        cursor = match.end()
-    output.append(text[cursor:])
-    return "".join(output)
-
-
-def create_surface_variant(
-    item: CanonicalItem,
-    family: SurfaceFamily,
-    rename_map: dict[str, str],
-) -> SurfaceVariant:
-    existing = item.variables + item.parameters
-    validate_rename_map(
-        rename_map,
-        existing_identifiers=existing,
-        scientific_constants=item.scientific_constants,
-    )
-    return SurfaceVariant(
-        family=family,
-        rename_map=dict(rename_map),
-        problem=apply_rename_map(item.problem, rename_map),
-        solution=apply_rename_map(item.solution, rename_map),
-    )
+    forbidden = expected | set(item.scientific_constants)
+    for replacement in rename_map.values():
+        if replacement in forbidden:
+            raise ValueError(f"replacement {replacement!r} collides with input symbols")
+        if not IDENTIFIER_RE.fullmatch(replacement):
+            raise ValueError(
+                f"replacement {replacement!r} violates the original prompt contract"
+            )
+    if not question.strip() or not solution.strip():
+        raise ValueError("surface question and solution must be non-empty")
 
 
 class SurfacePipeline:
@@ -128,90 +76,61 @@ class SurfacePipeline:
         self.proposer = proposer
         self.store = store
 
-    async def propose_map(
+    async def run_family(
         self,
         item: CanonicalItem,
         family: SurfaceFamily,
-    ) -> dict[str, str]:
-        symbols = [(value, "free variable") for value in item.variables] + [
-            (value, "fixed parameter") for value in item.parameters
-        ]
-        forbidden = list(
-            dict.fromkeys(
-                item.variables + item.parameters + item.scientific_constants
-            )
+    ) -> SurfaceVariant:
+        request_id = f"{item.item_id}.surface.{family}"
+        system_prompt = surface_system(family)
+        user_prompt = surface_user(item)
+        response = await self.proposer.generate_json(
+            system_prompt=system_prompt,
+            user_prompt=user_prompt,
+            request_id=request_id,
         )
-
-        async def propose(symbol: str, role: str) -> tuple[str, str]:
-            request_id = f"{item.item_id}.surface.{family}.{symbol}"
-            user_prompt = surface_name_user(
-                symbol=symbol,
-                role=role,
-                family=family,
-                context=(item.problem + "\n" + item.solution)[:12_000],
-                forbidden=forbidden,
-            )
-            response = await self.proposer.generate_json(
-                system_prompt=SURFACE_NAME_SYSTEM,
-                user_prompt=user_prompt,
+        self.store.write_call(
+            ModelCallRecord(
                 request_id=request_id,
+                model=response.model,
+                system_prompt=system_prompt,
+                user_prompt=user_prompt,
+                response_data=response.data,
+                raw_text=response.raw_text,
+                provider_response_id=response.response_id,
+                usage=response.usage,
             )
-            self.store.write_call(
-                ModelCallRecord(
-                    request_id=request_id,
-                    model=response.model,
-                    system_prompt=SURFACE_NAME_SYSTEM,
-                    user_prompt=user_prompt,
-                    response_data=response.data,
-                    raw_text=response.raw_text,
-                    provider_response_id=response.response_id,
-                    usage=response.usage,
-                )
-            )
-            replacement = str(response.data["replacement"])
-            return symbol, replacement
-
-        proposals = await asyncio.gather(
-            *(propose(symbol, role) for symbol, role in symbols)
         )
-        rename_map = dict(proposals)
-        validate_rename_map(
-            rename_map,
-            existing_identifiers=item.variables + item.parameters,
-            scientific_constants=item.scientific_constants,
+        rename_map = {
+            str(old): str(new)
+            for old, new in dict(response.data["map"]).items()
+        }
+        question = str(response.data["question"])
+        solution = str(response.data["solution"])
+        validate_surface_variant(
+            item,
+            rename_map=rename_map,
+            question=question,
+            solution=solution,
         )
-        self.store.write_stage(
-            f"surface_{family}_map",
-            {"rename_map": rename_map},
-            request_id=None,
+        variant = SurfaceVariant(
+            family=family,
+            rename_map=rename_map,
+            question=question,
+            solution=solution,
         )
-        return rename_map
-
-    async def run_family(
-        self,
-        item: CanonicalItem,
-        family: SurfaceFamily,
-    ) -> SurfaceVariant:
-        rename_map = await self.propose_map(item, family)
-        variant = create_surface_variant(item, family, rename_map)
         self.store.write_stage(
             f"surface_{family}_variant",
             variant.as_release_payload(),
-            request_id=None,
+            request_id=request_id,
         )
         return variant
 
-    async def run_all(self, item: CanonicalItem) -> dict[SurfaceFamily, SurfaceVariant]:
-        # Run families sequentially so the immutable call log stays easy to
-        # inspect; symbol proposals within each family remain concurrent.
+    async def run_all(
+        self,
+        item: CanonicalItem,
+    ) -> dict[SurfaceFamily, SurfaceVariant]:
         output: dict[SurfaceFamily, SurfaceVariant] = {}
         for family in SURFACE_FAMILIES:
             output[family] = await self.run_family(item, family)
         return output
-
-
-def deterministic_garbled_name(item_id: str, symbol: str, length: int = 12) -> str:
-    """Stable offline GS name for smoke tests; production follows the proposer."""
-
-    digest = hashlib.sha256(f"{item_id}:{symbol}".encode()).hexdigest()
-    return "v" + digest[: max(3, length - 1)]
diff --git a/tests/conftest.py b/tests/conftest.py
index 80ccc2e..ab3003b 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -22,105 +22,34 @@ def item() -> CanonicalItem:
 
 
 @pytest.fixture
-def dag_dict() -> dict:
-    return {
-        "nodes": [
-            {
-                "node_id": "n1",
-                "claim": "(a-1)^2 >= 0",
-                "derivation": "a square is nonnegative",
-                "dependencies": [],
-                "source_span": "Since (a-1)^2 >= 0",
-                "mathematical_objects": ["a", "(a-1)^2"],
-            },
-            {
-                "node_id": "n2",
-                "claim": "a+1/a >= 2",
-                "derivation": "expand and divide by positive a",
-                "dependencies": ["n1"],
-                "source_span": "Divide by a>0",
-                "mathematical_objects": ["a"],
-            },
-        ],
-        "terminal_node_id": "n2",
-    }
-
-
-@pytest.fixture
 def plan_dict() -> dict:
     return {
-        "steps": [
-            {
-                "node_id": "n1",
-                "method_label": "use nonnegativity of a square",
-                "input_roles": ["positive scalar"],
-                "output_role": "nonnegative expression",
-                "invariants": ["scalar is real"],
-            },
-            {
-                "node_id": "n2",
-                "method_label": "expand and divide by a positive quantity",
-                "input_roles": ["nonnegative expression", "positive scalar"],
-                "output_role": "target inequality",
-                "invariants": ["divisor is positive"],
-            },
+        "core_steps": [
+            "use nonnegativity of a square",
+            "expand and divide by a positive quantity",
         ],
-        "terminal_node_id": "n2",
+        "mutable_slots": {
+            "slot1": {
+                "description": "the positive reference value",
+                "original": "1",
+            }
+        },
     }
 
 
 @pytest.fixture
-def replacement_dict() -> dict:
+def candidate_dict() -> dict:
     return {
-        "target_node_id": "n1",
-        "original_object": "(a-1)^2",
-        "replacement_object": "(x-2)^2",
-        "guard_conditions": ["x>0"],
-        "guard_evidence": ["choose x in the positive reals"],
-        "expected_downstream_changes": ["expand around 2", "divide by x"],
-        "rationale": "the same square-nonnegativity plan applies",
-    }
-
-
-@pytest.fixture
-def diffused_dict(plan_dict: dict) -> dict:
-    return {
-        "node_instantiations": [
-            {
-                "node_id": "n1",
-                "method_label": plan_dict["steps"][0]["method_label"],
-                "instantiated_claim": "(x-2)^2 >= 0",
-                "instantiated_derivation": "a square is nonnegative",
-                "dependencies": [],
-            },
-            {
-                "node_id": "n2",
-                "method_label": plan_dict["steps"][1]["method_label"],
-                "instantiated_claim": "x+4/x >= 4",
-                "instantiated_derivation": "expand and divide by positive x",
-                "dependencies": ["n1"],
-            },
-        ],
-        "regenerated_proof": (
-            "Since (x-2)^2 >= 0, expand and divide by x>0 to get x+4/x >= 4."
+        "question": "Let x>0. Prove that x+4/x >= 4.",
+        "solution": (
+            "Since (x-2)^2 >= 0, expand and divide by x>0 "
+            "to get x+4/x >= 4."
         ),
-        "terminal_answer": "x+4/x >= 4",
-    }
-
-
-@pytest.fixture
-def candidate_dict(replacement_dict: dict, diffused_dict: dict) -> dict:
-    return {
-        "problem": "Let x>0. Prove that x+4/x >= 4.",
-        "proof": diffused_dict["regenerated_proof"],
-        "terminal_answer": diffused_dict["terminal_answer"],
-        "node_instantiations": diffused_dict["node_instantiations"],
-        "replacement": replacement_dict,
     }
 
 
 def changed_candidate(candidate: dict, suffix: str) -> dict:
     value = copy.deepcopy(candidate)
-    value["problem"] = value["problem"] + f" [{suffix}]"
-    value["proof"] = value["proof"] + f" Repair {suffix}."
+    value["question"] = value["question"] + f" [{suffix}]"
+    value["solution"] = value["solution"] + f" Repair {suffix}."
     return value
diff --git a/tests/test_models.py b/tests/test_models.py
index 2c678a0..71dc854 100644
--- a/tests/test_models.py
+++ b/tests/test_models.py
@@ -3,69 +3,38 @@ from __future__ import annotations
 import pytest
 from pydantic import ValidationError
 
-from gap_pipeline.models import (
-    DiffusedProof,
-    MethodPlan,
-    ProofDAG,
-    ReplacementSpec,
-)
+from gap_pipeline.models import JudgeVerdict, KernelCandidate, KernelPlan
 
 
-def test_valid_dag_and_plan(dag_dict: dict, plan_dict: dict) -> None:
-    dag = ProofDAG.model_validate(dag_dict)
-    assert [node.node_id for node in dag.topological_nodes()] == ["n1", "n2"]
-    assert dag.leaf_node_ids() == ["n1"]
-    plan = MethodPlan.model_validate(plan_dict)
-    plan.validate_against(dag)
+def test_valid_plan_and_candidate(plan_dict: dict, candidate_dict: dict) -> None:
+    plan = KernelPlan.model_validate(plan_dict)
+    candidate = KernelCandidate.model_validate(candidate_dict)
+    assert len(plan.core_steps) == 2
+    assert candidate.question.startswith("Let x")
 
 
-def test_cycle_is_rejected(dag_dict: dict) -> None:
-    dag_dict["nodes"][0]["dependencies"] = ["n2"]
-    with pytest.raises(ValidationError, match="cycle"):
-        ProofDAG.model_validate(dag_dict)
+def test_plan_requires_slots(plan_dict: dict) -> None:
+    plan_dict["mutable_slots"] = {}
+    with pytest.raises(ValidationError, match="mutable slot"):
+        KernelPlan.model_validate(plan_dict)
 
 
-def test_node_disconnected_from_terminal_is_rejected(dag_dict: dict) -> None:
-    dag_dict["nodes"].append(
-        {
-            "node_id": "orphan",
-            "claim": "unused claim",
-            "derivation": "unused derivation",
-            "dependencies": [],
-            "source_span": "",
-            "mathematical_objects": [],
-        }
-    )
-    with pytest.raises(ValidationError, match="disconnected"):
-        ProofDAG.model_validate(dag_dict)
+def test_plan_rejects_more_than_five_core_steps(plan_dict: dict) -> None:
+    plan_dict["core_steps"] = [f"step {index}" for index in range(6)]
+    with pytest.raises(ValidationError):
+        KernelPlan.model_validate(plan_dict)
 
 
-def test_method_reordering_is_rejected(dag_dict: dict, plan_dict: dict) -> None:
-    dag = ProofDAG.model_validate(dag_dict)
-    plan_dict["steps"].reverse()
-    plan = MethodPlan.model_validate(plan_dict)
-    with pytest.raises(ValueError, match="topological order"):
-        plan.validate_against(dag)
+def test_candidate_requires_question_and_solution(candidate_dict: dict) -> None:
+    candidate_dict["solution"] = ""
+    with pytest.raises(ValidationError, match="non-empty"):
+        KernelCandidate.model_validate(candidate_dict)
 
 
-def test_replacement_must_target_leaf(
-    dag_dict: dict, replacement_dict: dict
-) -> None:
-    dag = ProofDAG.model_validate(dag_dict)
-    replacement_dict["target_node_id"] = "n2"
-    replacement = ReplacementSpec.model_validate(replacement_dict)
-    with pytest.raises(ValueError, match="not a DAG leaf"):
-        replacement.validate_against(dag)
-
-
-def test_diffusion_cannot_change_dag_dependencies(
-    dag_dict: dict,
-    plan_dict: dict,
-    diffused_dict: dict,
-) -> None:
-    dag = ProofDAG.model_validate(dag_dict)
-    plan = MethodPlan.model_validate(plan_dict)
-    diffused_dict["node_instantiations"][1]["dependencies"] = []
-    diffused = DiffusedProof.model_validate(diffused_dict)
-    with pytest.raises(ValueError, match="changes dependencies"):
-        diffused.validate_against(dag, plan)
+def test_accept_verdict_cannot_report_blocking_issue() -> None:
+    with pytest.raises(ValidationError, match="cannot contain"):
+        JudgeVerdict(
+            verdict="accept",
+            step_by_step_check="checked",
+            blocking_issues="an error",
+        )
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index ea1077a..864e967 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -10,62 +10,56 @@ from gap_pipeline.store import RunStore
 from conftest import changed_candidate
 
 
-def accept_verdict() -> dict:
+def accept_review() -> dict[str, str]:
     return {
         "verdict": "accept",
-        "step_by_step_check": "n1 passes; n2 passes",
+        "step_by_step_check": "every method label is instantiated",
         "blocking_issues": "",
         "patch_suggestion": "",
     }
 
 
-def reject_verdict(iteration: int) -> dict:
+def reject_review(iteration: int) -> dict[str, str]:
     return {
         "verdict": "reject",
-        "step_by_step_check": f"n1 passes; n2 fails at iteration {iteration}",
-        "blocking_issues": "the terminal algebra needs correction",
-        "patch_suggestion": "correct the terminal algebra without changing the plan",
+        "step_by_step_check": f"terminal step fails at iteration {iteration}",
+        "blocking_issues": "terminal algebra is incorrect",
+        "patch_suggestion": "correct the terminal algebra",
     }
 
 
 def proposer_responses(
     item_id: str,
-    dag_dict: dict,
     plan_dict: dict,
-    replacement_dict: dict,
-    diffused_dict: dict,
     candidate_dict: dict,
 ) -> dict:
     return {
-        f"{item_id}.stage1": dag_dict,
-        f"{item_id}.stage2": plan_dict,
-        f"{item_id}.stage3": replacement_dict,
-        f"{item_id}.stage4": diffused_dict,
-        f"{item_id}.stage5": candidate_dict,
+        f"{item_id}.plan": plan_dict,
+        f"{item_id}.candidate": candidate_dict,
+    }
+
+
+def repair_response(candidate: dict) -> dict[str, str]:
+    return {
+        "corrected_question": candidate["question"],
+        "corrected_solution": candidate["solution"],
+        "changes_made": "corrected the terminal algebra",
     }
 
 
 def test_two_consecutive_unanimous_passes_accept(
     tmp_path,
     item,
-    dag_dict,
     plan_dict,
-    replacement_dict,
-    diffused_dict,
     candidate_dict,
 ) -> None:
     proposer = ScriptedClient(
-        proposer_responses(
-            item.item_id,
-            dag_dict,
-            plan_dict,
-            replacement_dict,
-            diffused_dict,
-            candidate_dict,
-        )
+        proposer_responses(item.item_id, plan_dict, candidate_dict)
     )
     judges = [
-        ScriptedClient({f"{item.item_id}.verify": [accept_verdict(), accept_verdict()]})
+        ScriptedClient(
+            {f"{item.item_id}.verify": [accept_review(), accept_review()]}
+        )
         for _ in range(5)
     ]
     pipeline = KernelPipeline(
@@ -78,138 +72,105 @@ def test_two_consecutive_unanimous_passes_accept(
     assert result.status == "accepted"
     assert len(result.iterations) == 2
     assert [row.pass_streak_after for row in result.iterations] == [1, 2]
-    assert not any(row.repaired for row in result.iterations)
 
     final_path = tmp_path / "items" / item.item_id / "final.json"
     final = json.loads(final_path.read_text())
-    assert final["human_audit_selected"] in {True, False}
-    assert len(list((final_path.parent / "calls").glob("*.json"))) == 15
+    assert final["accepted_candidate_sha256"]
+    assert len(list((final_path.parent / "calls").glob("*.json"))) == 12
 
 
 def test_rejection_resets_streak_and_repairs(
     tmp_path,
     item,
-    dag_dict,
     plan_dict,
-    replacement_dict,
-    diffused_dict,
     candidate_dict,
 ) -> None:
     repaired = changed_candidate(candidate_dict, "fixed")
-    responses = proposer_responses(
-        item.item_id,
-        dag_dict,
-        plan_dict,
-        replacement_dict,
-        diffused_dict,
-        candidate_dict,
-    )
-    responses[f"{item.item_id}.repair"] = repaired
+    responses = proposer_responses(item.item_id, plan_dict, candidate_dict)
+    responses[f"{item.item_id}.repair"] = repair_response(repaired)
     proposer = ScriptedClient(responses)
     judges = []
     for judge_id in range(1, 6):
-        first = reject_verdict(1) if judge_id == 1 else accept_verdict()
+        first = reject_review(1) if judge_id == 1 else accept_review()
         judges.append(
             ScriptedClient(
                 {
                     f"{item.item_id}.verify": [
                         first,
-                        accept_verdict(),
-                        accept_verdict(),
+                        accept_review(),
+                        accept_review(),
                     ]
                 }
             )
         )
-    pipeline = KernelPipeline(
-        proposer=proposer,
-        judges=judges,
-        store=RunStore(tmp_path, item.item_id),
-        config=PipelineConfig(proposer_model="scripted", judge_model="scripted"),
+    result = asyncio.run(
+        KernelPipeline(
+            proposer=proposer,
+            judges=judges,
+            store=RunStore(tmp_path, item.item_id),
+            config=PipelineConfig(
+                proposer_model="scripted",
+                judge_model="scripted",
+            ),
+        ).run(item)
     )
-    result = asyncio.run(pipeline.run(item))
     assert result.status == "accepted"
-    assert len(result.iterations) == 3
-    assert result.iterations[0].repaired
     assert [row.pass_streak_after for row in result.iterations] == [0, 1, 2]
-    assert (
-        result.iterations[0].repaired_candidate_sha256
-        == result.iterations[1].candidate_sha256
-        == result.iterations[2].candidate_sha256
-    )
+    assert result.iterations[0].candidate_sha256 != result.iterations[1].candidate_sha256
+    assert result.iterations[1].candidate_sha256 == result.iterations[2].candidate_sha256
 
 
-def test_rejection_after_one_pass_resets_streak_on_new_candidate(
+def test_rejection_after_one_pass_resets_streak(
     tmp_path,
     item,
-    dag_dict,
     plan_dict,
-    replacement_dict,
-    diffused_dict,
     candidate_dict,
 ) -> None:
     repaired = changed_candidate(candidate_dict, "after-broken-streak")
-    responses = proposer_responses(
-        item.item_id,
-        dag_dict,
-        plan_dict,
-        replacement_dict,
-        diffused_dict,
-        candidate_dict,
-    )
-    responses[f"{item.item_id}.repair"] = repaired
+    responses = proposer_responses(item.item_id, plan_dict, candidate_dict)
+    responses[f"{item.item_id}.repair"] = repair_response(repaired)
     proposer = ScriptedClient(responses)
     judges = []
     for judge_id in range(1, 6):
-        second = reject_verdict(2) if judge_id == 3 else accept_verdict()
+        second = reject_review(2) if judge_id == 3 else accept_review()
         judges.append(
             ScriptedClient(
                 {
                     f"{item.item_id}.verify": [
-                        accept_verdict(),
+                        accept_review(),
                         second,
-                        accept_verdict(),
-                        accept_verdict(),
+                        accept_review(),
+                        accept_review(),
                     ]
                 }
             )
         )
-    pipeline = KernelPipeline(
-        proposer=proposer,
-        judges=judges,
-        store=RunStore(tmp_path, item.item_id),
-        config=PipelineConfig(proposer_model="scripted", judge_model="scripted"),
+    result = asyncio.run(
+        KernelPipeline(
+            proposer=proposer,
+            judges=judges,
+            store=RunStore(tmp_path, item.item_id),
+            config=PipelineConfig(
+                proposer_model="scripted",
+                judge_model="scripted",
+            ),
+        ).run(item)
     )
-    result = asyncio.run(pipeline.run(item))
     assert result.status == "accepted"
     assert [row.pass_streak_after for row in result.iterations] == [1, 0, 1, 2]
     assert result.iterations[0].candidate_sha256 == result.iterations[1].candidate_sha256
-    assert result.iterations[1].repaired_candidate_sha256 != result.iterations[1].candidate_sha256
-    assert (
-        result.iterations[1].repaired_candidate_sha256
-        == result.iterations[2].candidate_sha256
-        == result.iterations[3].candidate_sha256
-    )
+    assert result.iterations[1].candidate_sha256 != result.iterations[2].candidate_sha256
 
 
 def test_fifteen_failed_rounds_reject(
     tmp_path,
     item,
-    dag_dict,
     plan_dict,
-    replacement_dict,
-    diffused_dict,
     candidate_dict,
 ) -> None:
-    responses = proposer_responses(
-        item.item_id,
-        dag_dict,
-        plan_dict,
-        replacement_dict,
-        diffused_dict,
-        candidate_dict,
-    )
+    responses = proposer_responses(item.item_id, plan_dict, candidate_dict)
     responses[f"{item.item_id}.repair"] = [
-        changed_candidate(candidate_dict, f"repair-{iteration}")
+        repair_response(changed_candidate(candidate_dict, f"repair-{iteration}"))
         for iteration in range(1, 15)
     ]
     proposer = ScriptedClient(responses)
@@ -217,20 +178,23 @@ def test_fifteen_failed_rounds_reject(
         ScriptedClient(
             {
                 f"{item.item_id}.verify": [
-                    reject_verdict(iteration) for iteration in range(1, 16)
+                    reject_review(iteration) for iteration in range(1, 16)
                 ]
             }
         )
         for _ in range(5)
     ]
-    pipeline = KernelPipeline(
-        proposer=proposer,
-        judges=judges,
-        store=RunStore(tmp_path, item.item_id),
-        config=PipelineConfig(proposer_model="scripted", judge_model="scripted"),
+    result = asyncio.run(
+        KernelPipeline(
+            proposer=proposer,
+            judges=judges,
+            store=RunStore(tmp_path, item.item_id),
+            config=PipelineConfig(
+                proposer_model="scripted",
+                judge_model="scripted",
+            ),
+        ).run(item)
     )
-    result = asyncio.run(pipeline.run(item))
     assert result.status == "rejected"
     assert len(result.iterations) == 15
-    assert sum(row.repaired for row in result.iterations) == 14
     assert all(row.pass_streak_after == 0 for row in result.iterations)
diff --git a/tests/test_prompts.py b/tests/test_prompts.py
new file mode 100644
index 0000000..a2bf892
--- /dev/null
+++ b/tests/test_prompts.py
@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import hashlib
+import inspect
+
+from gap_pipeline import prompts
+from gap_pipeline.clients import OpenAIJsonClient
+
+
+EXPECTED = {
+    "KERNEL_PLAN_SYSTEM": "0c817ede22fd407411ee588a02cf7a14ddcbb3d607aa4db6b5f81619609bcfef",
+    "KERNEL_PLAN_PROMPT": "4aafd1c25d67c5a433c8ad48dcf347cfe53436e81cfde1883e044ca3c17d855d",
+    "KERNEL_GENERATE_SYSTEM": "db0c902e15c9830a3c3860509bdb8a17b412d6b646b0f7e3278f32669046212b",
+    "KERNEL_GENERATE_PROMPT": "89c8a05dcda1b866b7835167c41184de9176ea058922e2cb250325825f79db92",
+    "JUDGE_SYSTEM_PROMPT": "50d5ee3ccbdf5196afea1eea60e61e74a324c89f14086b0ec6a8e14cb4ad0f30",
+    "JUDGE_USER_TEMPLATE": "59338f0a2522bda73006e7f62243578596cb8113da90fd8bfe5767d5a2bf4c2b",
+    "FIX_SYSTEM_PROMPT": "1c028afeac4bfbf92dd56cfce7b004b0823ab721f6f71e776bb90a5c652c1ed8",
+    "FIX_USER_TEMPLATE": "f12126acdc1518e872cd61f6baf2eedd49f1ad85eca69438e78128293959a8e9",
+    "SURFACE_SYSTEM_BASE": "d05ab24b4afd4ccd23e7954fa1111ce9c74893f8572c62617a3e568e3a743994",
+    "SURFACE_TASK_COMMON": "1cca2d06215207aa85ae36b368298a0b13ad4c2ef475079f1df0b2c4d9a5edd1",
+    "SURFACE_TASK_DESCRIPTIVE": "06b9b9c6a65ceae6818f30387cfd6e18bb46bbc1ccdf79adbba73f8849f6b354",
+    "SURFACE_TASK_CONFUSING": "7df3748ab186bbb6d9f421de5a7200754105154036d5cf6fb21ee32ac1f61848",
+    "SURFACE_TASK_MISLEADING": "fcde08ded42df91b44cb403f9a439550077245d71586adb6da9fb4983a4ea76e",
+    "SURFACE_TASK_GARBLED": "576301ca8e9eff2e4a2d9ed6a74f884a5a881ba22fe3a38e22e5eb46e341fac0",
+    "SURFACE_RETURN_SPEC": "b9f33abc9c8ab5e30b0154a6eebbd139d91afea6a744b454d4e150a6f7224a26",
+    "SURFACE_USER_TEMPLATE": "5d9043d7d1c1033db1aea0696812a2c4cbb6f3e28ca9d7b6436c036b40831fe7",
+}
+
+
+def test_prompt_values_are_byte_locked() -> None:
+    actual = {
+        name: hashlib.sha256(getattr(prompts, name).encode("utf-8")).hexdigest()
+        for name in EXPECTED
+    }
+    assert actual == EXPECTED
+
+
+def test_o3_adapter_does_not_send_temperature() -> None:
+    source = inspect.getsource(OpenAIJsonClient.generate_json)
+    assert '"temperature"' not in source
diff --git a/tests/test_release.py b/tests/test_release.py
index 3653ff1..b825478 100644
--- a/tests/test_release.py
+++ b/tests/test_release.py
@@ -70,7 +70,7 @@ def test_offline_release_export_verifies_and_assembles(
         (output_root / "records" / f"{item.item_id}.json").read_text()
     )
     assert set(output["variants"]) == {*SURFACE_FAMILIES, "kernel_variant"}
-    assert output["variants"]["kernel_variant"]["question"] == candidate.problem
+    assert output["variants"]["kernel_variant"]["question"] == candidate.question
     with pytest.raises(FileExistsError):
         export_release(
             source_dataset=source_dir,
diff --git a/tests/test_surface.py b/tests/test_surface.py
index 78c4d5e..4d92de3 100644
--- a/tests/test_surface.py
+++ b/tests/test_surface.py
@@ -1,43 +1,58 @@
 from __future__ import annotations
 
+import asyncio
+
 import pytest
 
-from gap_pipeline.surface import (
-    apply_rename_map,
-    create_surface_variant,
-    validate_rename_map,
-)
+from gap_pipeline.clients import ScriptedClient
+from gap_pipeline.store import RunStore
+from gap_pipeline.surface import SurfacePipeline, validate_surface_variant
 
 
-def test_math_only_rename_preserves_english_article(item) -> None:
-    text = r"Let \(a\) be a randomly chosen value with \(a>0\)."
-    renamed = apply_rename_map(text, {"a": "coefalpha"})
-    assert renamed == (
-        r"Let \(coefalpha\) be a randomly chosen value with \(coefalpha>0\)."
+def test_surface_pipeline_uses_full_original_contract(tmp_path, item) -> None:
+    response = {
+        "map": {"a": "positivequantity"},
+        "question": r"Let \(positivequantity>0\). Prove the renamed inequality.",
+        "solution": r"Apply the same proof to \(positivequantity\).",
+    }
+    client = ScriptedClient(
+        {f"{item.item_id}.surface.descriptive_long": response}
     )
-
-
-def test_surface_variant_changes_problem_and_solution(item) -> None:
-    variant = create_surface_variant(
-        item,
-        "descriptive_long",
-        {"a": "positiveScalar"},
+    variant = asyncio.run(
+        SurfacePipeline(
+            client,
+            RunStore(tmp_path, item.item_id),
+        ).run_family(item, "descriptive_long")
     )
-    assert "positiveScalar" in variant.problem
-    assert "positiveScalar" in variant.solution
-    assert "Let positiveScalar" not in variant.problem
-
-
-def test_collision_and_nonbijection_are_rejected(item) -> None:
-    with pytest.raises(ValueError, match="collides"):
-        validate_rename_map(
-            {"a": "a"},
-            existing_identifiers=["a"],
-            scientific_constants=[],
+    assert variant.rename_map == {"a": "positivequantity"}
+    assert variant.question == response["question"]
+    assert variant.solution == response["solution"]
+
+
+def test_missing_symbol_and_nonbijection_are_rejected(item) -> None:
+    with pytest.raises(ValueError, match="every var and param"):
+        validate_surface_variant(
+            item,
+            rename_map={},
+            question="question",
+            solution="solution",
         )
+
+    item.variables.append("b")
     with pytest.raises(ValueError, match="one-to-one"):
-        validate_rename_map(
-            {"a": "sharedName", "b": "sharedName"},
-            existing_identifiers=["a", "b"],
-            scientific_constants=[],
+        validate_surface_variant(
+            item,
+            rename_map={"a": "sharedname", "b": "sharedname"},
+            question="question",
+            solution="solution",
+        )
+
+
+def test_original_identifier_contract_is_enforced(item) -> None:
+    with pytest.raises(ValueError, match="original prompt contract"):
+        validate_surface_variant(
+            item,
+            rename_map={"a": "TooShort"},
+            question="question",
+            solution="solution",
         )