Restore original GAP prompts and lock prompt bytes

7 files changed, 494 insertions, 1025 deletions

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