Add minimal GAP reproduction package

11 files changed, 2499 insertions, 0 deletions

diff --git a/src/gap_pipeline/__init__.py b/src/gap_pipeline/__init__.py
new file mode 100644
index 0000000..2591afe
--- /dev/null
+++ b/src/gap_pipeline/__init__.py
@@ -0,0 +1,24 @@
+"""Reference implementation of GAP."""
+
+from .models import (
+    CanonicalItem,
+    KernelCandidate,
+    KernelRunResult,
+    MethodPlan,
+    ProofDAG,
+    ReplacementSpec,
+)
+from .pipeline import KernelPipeline, PipelineConfig
+
+__all__ = [
+    "CanonicalItem",
+    "KernelCandidate",
+    "KernelPipeline",
+    "KernelRunResult",
+    "MethodPlan",
+    "PipelineConfig",
+    "ProofDAG",
+    "ReplacementSpec",
+]
+
+__version__ = "0.1.0"
diff --git a/src/gap_pipeline/cli.py b/src/gap_pipeline/cli.py
new file mode 100644
index 0000000..105cc97
--- /dev/null
+++ b/src/gap_pipeline/cli.py
@@ -0,0 +1,168 @@
+"""Command-line interface for generation and all offline checks."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+
+from .clients import OpenAIJsonClient
+from .models import CanonicalItem
+from .offline import (
+    align_run_to_release,
+    load_dataset,
+    summarize_run,
+    validate_public_dataset,
+)
+from .pipeline import KernelPipeline, PipelineConfig
+from .release import export_release
+from .store import RunStore
+from .surface import SurfacePipeline
+
+
+def write_or_print(payload: dict[str, Any], output: Path | None) -> None:
+    text = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
+    if output is None:
+        print(text, end="")
+    else:
+        output.parent.mkdir(parents=True, exist_ok=True)
+        output.write_text(text, encoding="utf-8")
+
+
+async def generate_kernel(args: argparse.Namespace) -> None:
+    records = load_dataset(args.dataset)
+    if args.item_id not in records:
+        raise SystemExit(f"item ID {args.item_id!r} not found in {args.dataset}")
+    item = CanonicalItem.from_public_record(records[args.item_id])
+    config = PipelineConfig(
+        proposer_model=args.proposer_model,
+        judge_model=args.judge_model,
+    )
+    proposer = OpenAIJsonClient(args.proposer_model)
+    judges = [
+        OpenAIJsonClient(
+            args.judge_model,
+            seed=None if args.seed is None else args.seed + judge_id,
+        )
+        for judge_id in range(1, 6)
+    ]
+    pipeline = KernelPipeline(
+        proposer=proposer,
+        judges=judges,
+        store=RunStore(args.run_dir, item.item_id),
+        config=config,
+    )
+    result = await pipeline.run(item)
+    print(result.model_dump_json(indent=2))
+
+
+async def generate_surfaces(args: argparse.Namespace) -> None:
+    records = load_dataset(args.dataset)
+    if args.item_id not in records:
+        raise SystemExit(f"item ID {args.item_id!r} not found in {args.dataset}")
+    item = CanonicalItem.from_public_record(records[args.item_id])
+    store = RunStore(args.run_dir, item.item_id)
+    store.write_input(item)
+    store.write_config(
+        {
+            "protocol_name": "gap-surface-v1",
+            "proposer_model": args.proposer_model,
+        }
+    )
+    pipeline = SurfacePipeline(OpenAIJsonClient(args.proposer_model), store)
+    variants = await pipeline.run_all(item)
+    print(
+        json.dumps(
+            {
+                family: variant.as_release_payload()
+                for family, variant in variants.items()
+            },
+            ensure_ascii=False,
+            indent=2,
+        )
+    )
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="gap-reproduce")
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    validate = subparsers.add_parser("validate-data")
+    validate.add_argument("dataset", type=Path)
+    validate.add_argument("--output", type=Path)
+
+    summarize = subparsers.add_parser("summarize-run")
+    summarize.add_argument("run_dir", type=Path)
+    summarize.add_argument("--output", type=Path)
+
+    align = subparsers.add_parser("align-release")
+    align.add_argument("run_dir", type=Path)
+    align.add_argument("dataset", type=Path)
+    align.add_argument("--output", type=Path)
+
+    generate = subparsers.add_parser("generate-kernel")
+    generate.add_argument("--dataset", type=Path, required=True)
+    generate.add_argument("--item-id", required=True)
+    generate.add_argument("--run-dir", type=Path, required=True)
+    generate.add_argument("--proposer-model", default="o3")
+    generate.add_argument("--judge-model", default="o3")
+    generate.add_argument(
+        "--seed",
+        type=int,
+        help="optional provider seed; omitted by default for o3 compatibility",
+    )
+
+    surfaces = subparsers.add_parser("generate-surfaces")
+    surfaces.add_argument("--dataset", type=Path, required=True)
+    surfaces.add_argument("--item-id", required=True)
+    surfaces.add_argument("--run-dir", type=Path, required=True)
+    surfaces.add_argument("--proposer-model", default="o3")
+
+    export = subparsers.add_parser("export-release")
+    export.add_argument("--source-dataset", type=Path, required=True)
+    export.add_argument("--surface-run-dir", type=Path, required=True)
+    export.add_argument("--kernel-run-dir", type=Path, required=True)
+    export.add_argument("--output-root", type=Path, required=True)
+    export.add_argument("--allow-partial", action="store_true")
+    export.add_argument(
+        "--item-id",
+        action="append",
+        dest="item_ids",
+        help="export only this item ID; repeat to select multiple items",
+    )
+    return parser
+
+
+def main() -> None:
+    args = build_parser().parse_args()
+    if args.command == "validate-data":
+        write_or_print(validate_public_dataset(args.dataset), args.output)
+    elif args.command == "summarize-run":
+        write_or_print(summarize_run(args.run_dir), args.output)
+    elif args.command == "align-release":
+        write_or_print(
+            align_run_to_release(args.run_dir, args.dataset),
+            args.output,
+        )
+    elif args.command == "generate-kernel":
+        asyncio.run(generate_kernel(args))
+    elif args.command == "generate-surfaces":
+        asyncio.run(generate_surfaces(args))
+    elif args.command == "export-release":
+        manifest = export_release(
+            source_dataset=args.source_dataset,
+            surface_run_root=args.surface_run_dir,
+            kernel_run_root=args.kernel_run_dir,
+            output_root=args.output_root,
+            allow_partial=args.allow_partial,
+            item_ids=set(args.item_ids) if args.item_ids else None,
+        )
+        print(json.dumps(manifest, ensure_ascii=False, indent=2))
+    else:
+        raise AssertionError(args.command)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/src/gap_pipeline/clients.py b/src/gap_pipeline/clients.py
new file mode 100644
index 0000000..3b3e40f
--- /dev/null
+++ b/src/gap_pipeline/clients.py
@@ -0,0 +1,168 @@
+"""JSON LLM adapters. Offline tests use ReplayClient or ScriptedClient."""
+
+from __future__ import annotations
+
+import json
+from collections import defaultdict, deque
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Protocol
+
+
+@dataclass(frozen=True)
+class ModelResponse:
+    data: dict[str, Any]
+    raw_text: str
+    model: str
+    response_id: str | None = None
+    usage: dict[str, Any] = field(default_factory=dict)
+
+
+class JsonLLM(Protocol):
+    model: str
+
+    async def generate_json(
+        self,
+        *,
+        system_prompt: str,
+        user_prompt: str,
+        request_id: str,
+    ) -> ModelResponse: ...
+
+
+class OpenAIJsonClient:
+    """Optional OpenAI chat-completions adapter.
+
+    No API request is made until ``generate_json`` is awaited.
+    """
+
+    def __init__(
+        self,
+        model: str,
+        *,
+        api_key: str | None = None,
+        max_completion_tokens: int = 16_000,
+        seed: int | None = None,
+    ) -> None:
+        try:
+            from openai import AsyncOpenAI
+        except ImportError as exc:
+            raise RuntimeError(
+                "Install the optional API dependency: pip install -e '.[api]'"
+            ) from exc
+        self.model = model
+        self._client = AsyncOpenAI(api_key=api_key)
+        self.max_completion_tokens = max_completion_tokens
+        self.seed = seed
+
+    async def generate_json(
+        self,
+        *,
+        system_prompt: str,
+        user_prompt: str,
+        request_id: str,
+    ) -> ModelResponse:
+        kwargs: dict[str, Any] = {
+            "model": self.model,
+            "messages": [
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt},
+            ],
+            "response_format": {"type": "json_object"},
+            "max_completion_tokens": self.max_completion_tokens,
+        }
+        if self.seed is not None:
+            kwargs["seed"] = self.seed
+        response = await self._client.chat.completions.create(**kwargs)
+        raw = response.choices[0].message.content or "{}"
+        data = json.loads(raw)
+        usage = (
+            response.usage.model_dump()
+            if response.usage is not None and hasattr(response.usage, "model_dump")
+            else {}
+        )
+        return ModelResponse(
+            data=data,
+            raw_text=raw,
+            model=self.model,
+            response_id=response.id,
+            usage=usage,
+        )
+
+
+class ReplayClient:
+    """Replay archived call responses keyed by request ID."""
+
+    def __init__(self, path: Path, model: str = "replay") -> None:
+        self.model = model
+        payload = json.loads(path.read_text(encoding="utf-8"))
+        if isinstance(payload, list):
+            self._responses = {
+                str(row["request_id"]): row["response_data"] for row in payload
+            }
+        elif isinstance(payload, dict):
+            self._responses = payload
+        else:
+            raise ValueError("replay file must be a JSON object or list")
+
+    async def generate_json(
+        self,
+        *,
+        system_prompt: str,
+        user_prompt: str,
+        request_id: str,
+    ) -> ModelResponse:
+        if request_id not in self._responses:
+            raise KeyError(f"no replay response for {request_id}")
+        data = self._responses[request_id]
+        return ModelResponse(
+            data=data,
+            raw_text=json.dumps(data, ensure_ascii=False),
+            model=self.model,
+            response_id=f"replay:{request_id}",
+        )
+
+
+class ScriptedClient:
+    """Small deterministic client for tests and offline smoke runs."""
+
+    def __init__(
+        self,
+        responses: dict[str, list[dict[str, Any]] | dict[str, Any]],
+        model: str = "scripted",
+    ) -> None:
+        self.model = model
+        self.calls: list[str] = []
+        self._responses: dict[str, deque[dict[str, Any]]] = {}
+        for prefix, values in responses.items():
+            sequence = values if isinstance(values, list) else [values]
+            self._responses[prefix] = deque(sequence)
+        self._prefix_counts: defaultdict[str, int] = defaultdict(int)
+
+    async def generate_json(
+        self,
+        *,
+        system_prompt: str,
+        user_prompt: str,
+        request_id: str,
+    ) -> ModelResponse:
+        self.calls.append(request_id)
+        prefixes = sorted(
+            (prefix for prefix in self._responses if request_id.startswith(prefix)),
+            key=len,
+            reverse=True,
+        )
+        if not prefixes:
+            raise KeyError(f"no scripted response matches {request_id}")
+        prefix = prefixes[0]
+        queue = self._responses[prefix]
+        if not queue:
+            raise IndexError(f"scripted responses exhausted for {prefix}")
+        data = queue.popleft()
+        self._prefix_counts[prefix] += 1
+        return ModelResponse(
+            data=data,
+            raw_text=json.dumps(data, ensure_ascii=False),
+            model=self.model,
+            response_id=f"scripted:{prefix}:{self._prefix_counts[prefix]}",
+        )
diff --git a/src/gap_pipeline/e2e.py b/src/gap_pipeline/e2e.py
new file mode 100644
index 0000000..093068f
--- /dev/null
+++ b/src/gap_pipeline/e2e.py
@@ -0,0 +1,301 @@
+"""End-to-end helpers used by the reproducibility notebook."""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+from typing import Any
+
+from .clients import OpenAIJsonClient, ScriptedClient
+from .models import CanonicalItem
+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
+
+
+def _ensure_fresh(path: Path) -> None:
+    if path.exists() and any(path.iterdir()):
+        raise FileExistsError(f"refusing to reuse non-empty output directory {path}")
+
+
+def _accept_verdict() -> dict[str, Any]:
+    return {
+        "verdict": "accept",
+        "step_by_step_check": "n1 and n2 instantiate the fixed method plan",
+        "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,
+    }
+    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,
+    }
+
+
+async def run_live_item(
+    *,
+    dataset_dir: Path,
+    item_id: str,
+    work_root: Path,
+    model: str = "o3",
+    api_key: str | None = None,
+) -> dict[str, Any]:
+    """Generate all five GAP variants for one Putnam item and export them."""
+
+    if not (api_key or os.getenv("OPENAI_API_KEY")):
+        raise RuntimeError(
+            "OPENAI_API_KEY is missing. Set it in the environment or enter it "
+            "with getpass in the notebook."
+        )
+    _ensure_fresh(work_root)
+    work_root.mkdir(parents=True, exist_ok=True)
+    surface_root = work_root / "surface-runs"
+    kernel_root = work_root / "kernel-runs"
+    release_root = work_root / "release"
+
+    records = load_dataset(dataset_dir)
+    if item_id not in records:
+        raise KeyError(f"{item_id!r} not found in {dataset_dir}")
+    item = CanonicalItem.from_public_record(records[item_id])
+
+    started = time.monotonic()
+    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,
+        }
+    )
+    surface_pipeline = SurfacePipeline(
+        OpenAIJsonClient(model, api_key=api_key),
+        surface_store,
+    )
+    surfaces = await surface_pipeline.run_all(item)
+
+    config = PipelineConfig(proposer_model=model, judge_model=model)
+    kernel_pipeline = 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)
+    if kernel.status != "accepted":
+        raise RuntimeError(
+            f"kernel candidate was rejected after {len(kernel.iterations)} rounds; "
+            f"inspect {kernel_root / 'items' / item.item_id}"
+        )
+
+    manifest = export_release(
+        source_dataset=dataset_dir,
+        surface_run_root=surface_root,
+        kernel_run_root=kernel_root,
+        output_root=release_root,
+        item_ids={item_id},
+    )
+    return {
+        "mode": "live",
+        "item_id": item_id,
+        "model": model,
+        "elapsed_seconds": round(time.monotonic() - started, 2),
+        "surface_families": sorted(surfaces),
+        "kernel_status": kernel.status,
+        "verification_rounds": len(kernel.iterations),
+        "export_status": manifest["status"],
+        "exported_item_count": manifest["exported_item_count"],
+        "work_root": str(work_root.resolve()),
+        "release_record": manifest["exported"][0]["path"],
+        "manifest": str((release_root / "manifest.json").resolve()),
+    }
+
+
+async def run_offline_smoke(work_root: Path) -> dict[str, Any]:
+    """Exercise the same end-to-end code path with deterministic responses."""
+
+    _ensure_fresh(work_root)
+    work_root.mkdir(parents=True, exist_ok=True)
+    source_root = work_root / "source"
+    source_root.mkdir()
+    surface_root = work_root / "surface-runs"
+    kernel_root = work_root / "kernel-runs"
+    release_root = work_root / "release"
+
+    fixture = _offline_fixture()
+    record = fixture["record"]
+    item_id = str(record["index"])
+    (source_root / f"{item_id}.json").write_text(
+        json.dumps(record, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    item = CanonicalItem.from_public_record(record)
+
+    surface_responses = {
+        f"{item_id}.surface.{family}.a": {
+            "replacement": {
+                "descriptive_long": "positiveQuantity",
+                "descriptive_long_confusing": "walnutVioletTerrace",
+                "descriptive_long_misleading": "primeFieldOrder",
+                "garbled_string": "xcQ7h2ZfRw9v",
+            }[family]
+        }
+        for family in SURFACE_FAMILIES
+    }
+    surface_store = RunStore(surface_root, item_id)
+    surface_store.write_input(item)
+    surface_store.write_config(
+        {"protocol_name": "gap-surface-smoke", "proposer_model": "scripted"}
+    )
+    surfaces = await SurfacePipeline(
+        ScriptedClient(surface_responses),
+        surface_store,
+    ).run_all(item)
+
+    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"],
+        }
+    )
+    judges = [
+        ScriptedClient(
+            {
+                f"{item_id}.verify": [
+                    _accept_verdict(),
+                    _accept_verdict(),
+                ]
+            }
+        )
+        for _ in range(5)
+    ]
+    kernel = await KernelPipeline(
+        proposer=proposer,
+        judges=judges,
+        store=RunStore(kernel_root, item_id),
+        config=PipelineConfig(
+            proposer_model="scripted",
+            judge_model="scripted",
+        ),
+    ).run(item)
+    manifest = export_release(
+        source_dataset=source_root,
+        surface_run_root=surface_root,
+        kernel_run_root=kernel_root,
+        output_root=release_root,
+        item_ids={item_id},
+    )
+    return {
+        "mode": "offline-smoke",
+        "item_id": item_id,
+        "surface_families": sorted(surfaces),
+        "kernel_status": kernel.status,
+        "verification_rounds": len(kernel.iterations),
+        "export_status": manifest["status"],
+        "exported_item_count": manifest["exported_item_count"],
+        "work_root": str(work_root.resolve()),
+        "release_record": manifest["exported"][0]["path"],
+        "manifest": str((release_root / "manifest.json").resolve()),
+    }
diff --git a/src/gap_pipeline/models.py b/src/gap_pipeline/models.py
new file mode 100644
index 0000000..fe7aeb7
--- /dev/null
+++ b/src/gap_pipeline/models.py
@@ -0,0 +1,294 @@
+"""Validated schemas for every GAP pipeline boundary."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+
+SCHEMA_VERSION = "gap-reference-v1"
+
+
+class StrictModel(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+
+class CanonicalItem(StrictModel):
+    item_id: str
+    problem: str
+    solution: str
+    problem_type: Literal["proof", "calculation"] | str = "proof"
+    variables: list[str] = Field(default_factory=list)
+    parameters: list[str] = Field(default_factory=list)
+    scientific_constants: list[str] = Field(default_factory=list)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+    @classmethod
+    def from_public_record(cls, record: dict[str, Any]) -> "CanonicalItem":
+        excluded = {
+            "index",
+            "question",
+            "solution",
+            "problem_type",
+            "vars",
+            "params",
+            "sci_consts",
+            "variants",
+        }
+        return cls(
+            item_id=str(record["index"]),
+            problem=str(record["question"]),
+            solution=str(record["solution"]),
+            problem_type=str(record.get("problem_type", "proof")),
+            variables=list(record.get("vars", [])),
+            parameters=list(record.get("params", [])),
+            scientific_constants=list(record.get("sci_consts", [])),
+            metadata={key: value for key, value in record.items() if key not in excluded},
+        )
+
+
+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 ProofDAG(StrictModel):
+    nodes: list[ProofNode]
+    terminal_node_id: str
+
+    @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)}"
+            )
+        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
+
+    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)
+
+
+class JudgeVerdict(StrictModel):
+    verdict: Literal["accept", "reject"]
+    step_by_step_check: str
+    blocking_issues: str = ""
+    patch_suggestion: str = ""
+
+    @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")
+        return self
+
+
+class IterationRecord(StrictModel):
+    iteration: int
+    candidate_sha256: str
+    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"]
+    accepted_candidate: KernelCandidate | None = None
+    iterations: list[IterationRecord]
+    rejection_reason: str = ""
+    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")
+        if self.status == "rejected" and not self.rejection_reason:
+            raise ValueError("rejected run must state a reason")
+        return self
+
+
+class ModelCallRecord(StrictModel):
+    request_id: str
+    model: str
+    system_prompt: str
+    user_prompt: str
+    response_data: dict[str, Any]
+    raw_text: str
+    provider_response_id: str | None = None
+    usage: dict[str, Any] = Field(default_factory=dict)
+    created_at: str = Field(
+        default_factory=lambda: datetime.now(timezone.utc).isoformat()
+    )
+
+
+class StageArtifact(StrictModel):
+    item_id: str
+    stage: str
+    payload_sha256: str
+    payload: dict[str, Any]
+    request_id: str | None = None
+    schema_version: str = SCHEMA_VERSION
+    created_at: str = Field(
+        default_factory=lambda: datetime.now(timezone.utc).isoformat()
+    )
diff --git a/src/gap_pipeline/offline.py b/src/gap_pipeline/offline.py
new file mode 100644
index 0000000..2be8922
--- /dev/null
+++ b/src/gap_pipeline/offline.py
@@ -0,0 +1,279 @@
+"""No-API validation, released-data alignment, and run statistics."""
+
+from __future__ import annotations
+
+import json
+import re
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+from .store import sha256_payload
+from .surface import IDENTIFIER_RE, SURFACE_FAMILIES
+
+
+REQUIRED_TOP_LEVEL = {
+    "index",
+    "question",
+    "solution",
+    "vars",
+    "params",
+    "sci_consts",
+    "variants",
+    "problem_type",
+}
+
+
+def load_dataset(dataset_dir: Path) -> dict[str, dict[str, Any]]:
+    records: dict[str, dict[str, Any]] = {}
+    for path in sorted(dataset_dir.glob("*.json")):
+        record = json.loads(path.read_text(encoding="utf-8"))
+        item_id = str(record.get("index", ""))
+        if not item_id:
+            raise ValueError(f"{path}: missing index")
+        if item_id in records:
+            raise ValueError(f"duplicate dataset ID {item_id}")
+        records[item_id] = record
+    return records
+
+
+def normalize_latex_symbol(value: str) -> str:
+    """Normalize notation aliases used in the released rename maps.
+
+    The public records sometimes contain both ``x_n`` and ``x_{n}``, or use a
+    different number of escaped backslashes between the symbol inventory and a
+    map key.  These are representation aliases, not two mathematical symbols.
+    This normalization is deliberately used only for diagnostics; it never
+    rewrites released text.
+    """
+
+    normalized = re.sub(r"\\\\+", r"\\", str(value))
+    normalized = re.sub(r"\s+", "", normalized)
+    previous = None
+    while previous != normalized:
+        previous = normalized
+        normalized = re.sub(r"_\{([^{}]+)\}", r"_\1", normalized)
+    return normalized
+
+
+def validate_public_dataset(dataset_dir: Path) -> dict[str, Any]:
+    records = load_dataset(dataset_dir)
+    errors: list[str] = []
+    warnings: list[str] = []
+    family_counts: Counter[str] = Counter()
+    kernel_metadata_counts: Counter[str] = Counter()
+    potential_target_collision_count = 0
+    unchanged_mapping_count = 0
+    tex_target_count = 0
+    alias_source_count = 0
+    source_outside_inventory_count = 0
+
+    for item_id, record in records.items():
+        missing = REQUIRED_TOP_LEVEL - set(record)
+        if missing:
+            errors.append(f"{item_id}: missing top-level fields {sorted(missing)}")
+            continue
+        if not record["question"].strip() or not record["solution"].strip():
+            errors.append(f"{item_id}: empty canonical question or solution")
+        expected_symbols = set(record.get("vars", [])) | set(record.get("params", []))
+        normalized_expected = {
+            normalize_latex_symbol(value) for value in expected_symbols
+        }
+        variants = record["variants"]
+
+        for family in SURFACE_FAMILIES:
+            family_counts[family] += int(family in variants)
+            if family not in variants:
+                errors.append(f"{item_id}: missing surface family {family}")
+                continue
+            variant = variants[family]
+            missing_variant = {"map", "question", "solution"} - set(variant)
+            if missing_variant:
+                errors.append(
+                    f"{item_id}/{family}: missing fields {sorted(missing_variant)}"
+                )
+                continue
+            rename_map = variant["map"]
+            canonical_sources: dict[str, str] = {}
+            missing_sources: list[str] = []
+            for source in rename_map:
+                normalized_source = normalize_latex_symbol(source)
+                canonical_sources[source] = normalized_source
+                if source not in expected_symbols and normalized_source in normalized_expected:
+                    alias_source_count += 1
+                elif normalized_source not in normalized_expected:
+                    missing_sources.append(source)
+                    source_outside_inventory_count += 1
+            if missing_sources:
+                warnings.append(
+                    f"{item_id}/{family}: map sources absent from normalized "
+                    f"symbol inventory {sorted(missing_sources)}"
+                )
+
+            # Alias keys such as x_n and x_{n} are one source.  A collision is
+            # counted only if two distinct normalized sources share a target.
+            target_to_sources: dict[str, set[str]] = {}
+            for source, replacement in rename_map.items():
+                target_to_sources.setdefault(
+                    normalize_latex_symbol(replacement), set()
+                ).add(canonical_sources[source])
+                if normalize_latex_symbol(source) == normalize_latex_symbol(
+                    replacement
+                ):
+                    unchanged_mapping_count += 1
+                if not IDENTIFIER_RE.fullmatch(str(replacement)):
+                    tex_target_count += 1
+                if not str(replacement).strip():
+                    errors.append(f"{item_id}/{family}: empty replacement target")
+            potential_target_collision_count += sum(
+                len(sources) - 1
+                for sources in target_to_sources.values()
+                if len(sources) > 1
+            )
+
+        if "kernel_variant" not in variants:
+            errors.append(f"{item_id}: missing kernel_variant")
+        else:
+            family_counts["kernel_variant"] += 1
+            kernel = variants["kernel_variant"]
+            if not str(kernel.get("question", "")).strip():
+                errors.append(f"{item_id}/kernel_variant: empty question")
+            if not str(kernel.get("solution", "")).strip():
+                errors.append(f"{item_id}/kernel_variant: empty solution")
+            if "_meta" in kernel:
+                kernel_metadata_counts["_meta"] += 1
+            if "metadata" in kernel:
+                kernel_metadata_counts["metadata"] += 1
+            if "_replacement_note" in kernel:
+                kernel_metadata_counts["_replacement_note"] += 1
+    return {
+        "dataset_dir": str(dataset_dir.resolve()),
+        "record_count": len(records),
+        "expected_record_count": 1051,
+        "record_count_matches": len(records) == 1051,
+        "family_counts": dict(family_counts),
+        "kernel_metadata_shapes": dict(kernel_metadata_counts),
+        "surface_map_checks": {
+            "potential_target_collision_count": potential_target_collision_count,
+            "unchanged_mapping_count": unchanged_mapping_count,
+            "tex_or_structured_target_count": tex_target_count,
+            "notation_alias_source_count": alias_source_count,
+            "source_outside_normalized_inventory_count": (
+                source_outside_inventory_count
+            ),
+            "interpretation": (
+                "TeX/structured targets, notation aliases, and potential "
+                "target collisions are descriptive diagnostics, not validation "
+                "errors. Strict rules apply only to newly generated maps."
+            ),
+        },
+        "errors": errors,
+        "warnings": warnings,
+        "valid": not errors,
+    }
+
+
+def normalize_text(value: str) -> str:
+    return re.sub(r"\s+", " ", value).strip()
+
+
+def summarize_run(run_dir: Path) -> dict[str, Any]:
+    finals = sorted(run_dir.glob("items/*/final.json"))
+    counts: Counter[str] = Counter()
+    iterations: list[int] = []
+    first_round_failed = 0
+    repaired_accepts = 0
+    cap_reached = 0
+    accepted_ids: list[str] = []
+    rejected_ids: list[str] = []
+
+    for path in finals:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+        status = str(payload["status"])
+        counts[status] += 1
+        item_id = str(payload["item_id"])
+        history = payload.get("iterations", [])
+        iterations.append(len(history))
+        if history and not history[0].get("unanimous", False):
+            first_round_failed += 1
+        if status == "accepted":
+            accepted_ids.append(item_id)
+            if any(row.get("repaired", False) for row in history):
+                repaired_accepts += 1
+        else:
+            rejected_ids.append(item_id)
+            if len(history) >= 15:
+                cap_reached += 1
+
+    return {
+        "run_dir": str(run_dir.resolve()),
+        "candidate_count": len(finals),
+        "status_counts": dict(counts),
+        "failed_first_unanimous_vote": first_round_failed,
+        "accepted_after_repair": repaired_accepts,
+        "finally_rejected": counts["rejected"],
+        "reached_iteration_cap": cap_reached,
+        "iteration_min": min(iterations) if iterations else None,
+        "iteration_max": max(iterations) if iterations else None,
+        "iteration_mean": sum(iterations) / len(iterations) if iterations else None,
+        "accepted_ids": accepted_ids,
+        "rejected_ids": rejected_ids,
+    }
+
+
+def align_run_to_release(run_dir: Path, dataset_dir: Path) -> dict[str, Any]:
+    release = load_dataset(dataset_dir)
+    rows = []
+    for path in sorted(run_dir.glob("items/*/final.json")):
+        payload = json.loads(path.read_text(encoding="utf-8"))
+        if payload.get("status") != "accepted":
+            continue
+        item_id = str(payload["item_id"])
+        candidate = payload["accepted_candidate"]
+        if item_id not in release:
+            rows.append({"item_id": item_id, "status": "missing_from_release"})
+            continue
+        released_kernel = release[item_id]["variants"]["kernel_variant"]
+        question_exact = candidate["problem"] == released_kernel["question"]
+        solution_exact = candidate["proof"] == released_kernel["solution"]
+        question_normalized = normalize_text(candidate["problem"]) == normalize_text(
+            released_kernel["question"]
+        )
+        solution_normalized = normalize_text(candidate["proof"]) == normalize_text(
+            released_kernel["solution"]
+        )
+        rows.append(
+            {
+                "item_id": item_id,
+                "status": (
+                    "exact_match"
+                    if question_exact and solution_exact
+                    else "normalized_match"
+                    if question_normalized and solution_normalized
+                    else "mismatch"
+                ),
+                "question_exact": question_exact,
+                "solution_exact": solution_exact,
+                "question_normalized": question_normalized,
+                "solution_normalized": solution_normalized,
+                "candidate_sha256": sha256_payload(
+                    {
+                        "question": candidate["problem"],
+                        "solution": candidate["proof"],
+                    }
+                ),
+                "release_sha256": sha256_payload(
+                    {
+                        "question": released_kernel["question"],
+                        "solution": released_kernel["solution"],
+                    }
+                ),
+            }
+        )
+    return {
+        "run_dir": str(run_dir.resolve()),
+        "dataset_dir": str(dataset_dir.resolve()),
+        "checked": len(rows),
+        "status_counts": dict(Counter(row["status"] for row in rows)),
+        "rows": rows,
+    }
diff --git a/src/gap_pipeline/pipeline.py b/src/gap_pipeline/pipeline.py
new file mode 100644
index 0000000..2c0deb4
--- /dev/null
+++ b/src/gap_pipeline/pipeline.py
@@ -0,0 +1,459 @@
+"""Five-stage kernel generation and the J=5, K=2, T=15 verifier."""
+
+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,
+    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,
+    judge_user,
+    method_user,
+    question_user,
+    repair_user,
+    replacement_user,
+)
+from .store import RunStore, sha256_payload
+
+
+class PipelineConfig(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    protocol_name: str = "gap-J5-K2-T15"
+    proposer_model: str
+    judge_model: str
+    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":
+        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%")
+        return self
+
+
+class KernelPipeline:
+    def __init__(
+        self,
+        *,
+        proposer: JsonLLM,
+        judges: list[JsonLLM],
+        store: RunStore,
+        config: PipelineConfig,
+    ) -> None:
+        if len(judges) != config.judge_count:
+            raise ValueError(
+                f"expected {config.judge_count} judge clients, received {len(judges)}"
+            )
+        self.proposer = proposer
+        self.judges = judges
+        self.store = store
+        self.config = config
+
+    async def _call(
+        self,
+        client: JsonLLM,
+        *,
+        request_id: str,
+        system_prompt: str,
+        user_prompt: str,
+        response_type: type[BaseModel],
+    ) -> BaseModel:
+        response = await client.generate_json(
+            system_prompt=system_prompt,
+            user_prompt=user_prompt,
+            request_id=request_id,
+        )
+        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,
+            )
+        )
+        return response_type.model_validate(response.data)
+
+    async def construct_dag(self, item: CanonicalItem) -> ProofDAG:
+        request_id = f"{item.item_id}.stage1.dag"
+        dag = await self._call(
+            self.proposer,
+            request_id=request_id,
+            system_prompt=DAG_SYSTEM,
+            user_prompt=dag_user(item),
+            response_type=ProofDAG,
+        )
+        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,
+            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,
+            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,
+            request_id=request_id,
+        )
+        return replacement
+
+    async def diffuse_dag(
+        self,
+        item: CanonicalItem,
+        dag: ProofDAG,
+        plan: MethodPlan,
+        replacement: ReplacementSpec,
+    ) -> DiffusedProof:
+        request_id = f"{item.item_id}.stage4.diffusion"
+        diffused = await self._call(
+            self.proposer,
+            request_id=request_id,
+            system_prompt=DIFFUSION_SYSTEM,
+            user_prompt=diffusion_user(item, dag, plan, replacement),
+            response_type=DiffusedProof,
+        )
+        assert isinstance(diffused, DiffusedProof)
+        diffused.validate_against(dag, plan)
+        self.store.write_stage(
+            "04_diffused_proof",
+            diffused,
+            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,
+            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,
+        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(
+            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,
+        )
+        assert isinstance(verdict, JudgeVerdict)
+        return verdict
+
+    async def _repair(
+        self,
+        *,
+        item: CanonicalItem,
+        dag: ProofDAG,
+        plan: MethodPlan,
+        candidate: KernelCandidate,
+        verdicts: list[JudgeVerdict],
+        iteration: int,
+    ) -> KernelCandidate:
+        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(
+            self.proposer,
+            request_id=request_id,
+            system_prompt=REPAIR_SYSTEM,
+            user_prompt=repair_user(item, dag, plan, candidate, issues),
+            response_type=KernelCandidate,
+        )
+        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,
+            request_id=request_id,
+        )
+        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,
+        candidate: KernelCandidate,
+    ) -> KernelRunResult:
+        iterations: list[IterationRecord] = []
+        pass_streak = 0
+        current = candidate
+        streak_candidate_sha: str | None = None
+
+        for iteration in range(1, self.config.max_iterations + 1):
+            current_sha = sha256_payload(current)
+            verdicts = list(
+                await asyncio.gather(
+                    *(
+                        self._judge_once(
+                            judge,
+                            item=item,
+                            plan=plan,
+                            candidate=current,
+                            iteration=iteration,
+                            judge_id=judge_id,
+                        )
+                        for judge_id, judge in enumerate(self.judges, start=1)
+                    )
+                )
+            )
+            unanimous = all(verdict.verdict == "accept" for verdict in verdicts)
+
+            if unanimous:
+                if streak_candidate_sha not in {None, current_sha}:
+                    raise AssertionError("pass streak crossed candidate versions")
+                streak_candidate_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
+
+            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,
+            )
+            iterations.append(record)
+            self.store.write_iteration(iteration, record)
+            current = repaired
+
+        result = KernelRunResult(
+            item_id=item.item_id,
+            status="rejected",
+            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,
+            }
+        )
+        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)
+        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
new file mode 100644
index 0000000..667a47a
--- /dev/null
+++ b/src/gap_pipeline/prompts.py
@@ -0,0 +1,338 @@
+"""Prompts and strict JSON contracts for the GAP pipeline."""
+
+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."""
+
+
+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)}
+
+GUARDED REPLACEMENT:
+{replacement.model_dump_json(indent=2)}
+
+Return:
+{{
+  "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"
+}}
+
+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."""
+
+
+def question_user(
+    item: CanonicalItem,
+    dag: ProofDAG,
+    plan: MethodPlan,
+    replacement: ReplacementSpec,
+    diffused_payload: dict,
+) -> str:
+    return f"""SOURCE PROBLEM (style reference only):
+{item.problem}
+
+FIXED METHOD PLAN:
+{plan.model_dump_json(indent=2)}
+
+REPLACEMENT:
+{replacement.model_dump_json(indent=2)}
+
+REGENERATED PROOF:
+{json.dumps(diffused_payload, ensure_ascii=False, indent=2)}
+
+Return:
+{{
+  "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()}
+}}
+
+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).
+
+Your job is verification, not solving. You receive:
+- the ORIGINAL problem statement and reference solution,
+- the abstract METHOD-LABEL sequence,
+- 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."""
+
+
+def judge_user(
+    item: CanonicalItem,
+    plan: MethodPlan,
+    candidate: KernelCandidate,
+    *,
+    judge_id: int,
+    iteration: int,
+) -> str:
+    return f"""JUDGE REPLICATE: {judge_id}
+ITERATION: {iteration}
+
+ORIGINAL PROBLEM:
+{item.problem}
+
+ORIGINAL REFERENCE SOLUTION:
+{item.solution}
+
+METHOD-LABEL SEQUENCE:
+{plan.model_dump_json(indent=2)}
+
+SLOT REPLACEMENT:
+{candidate.replacement.model_dump_json(indent=2)}
+
+CANDIDATE VARIANT 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)}
+
+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."""
+
+
+def repair_user(
+    item: CanonicalItem,
+    dag: ProofDAG,
+    plan: MethodPlan,
+    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."""
+
+
+def surface_name_user(
+    *,
+    symbol: str,
+    role: str,
+    family: str,
+    context: str,
+    forbidden: list[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"}}."""
diff --git a/src/gap_pipeline/release.py b/src/gap_pipeline/release.py
new file mode 100644
index 0000000..f86ce64
--- /dev/null
+++ b/src/gap_pipeline/release.py
@@ -0,0 +1,138 @@
+"""Offline assembly of a GAP release from validated run artifacts."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+from .offline import load_dataset
+from .store import safe_name, sha256_payload
+from .surface import SURFACE_FAMILIES
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for block in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def read_stage(run_root: Path, item_id: str, stage: str) -> dict[str, Any]:
+    path = (
+        run_root
+        / "items"
+        / safe_name(item_id)
+        / "stages"
+        / f"{safe_name(stage)}.json"
+    )
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if payload.get("payload_sha256") != sha256_payload(payload.get("payload")):
+        raise ValueError(f"stage digest mismatch: {path}")
+    return payload["payload"]
+
+
+def read_kernel(run_root: Path, item_id: str) -> tuple[dict[str, Any], Path]:
+    path = run_root / "items" / safe_name(item_id) / "final.json"
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if payload.get("status") != "accepted" or not payload.get("accepted_candidate"):
+        raise ValueError(f"kernel run is not accepted: {path}")
+    candidate = payload["accepted_candidate"]
+    expected = payload.get("accepted_candidate_sha256")
+    if expected and expected != sha256_payload(candidate):
+        raise ValueError(f"accepted candidate digest mismatch: {path}")
+    return payload, path
+
+
+def export_release(
+    *,
+    source_dataset: Path,
+    surface_run_root: Path,
+    kernel_run_root: Path,
+    output_root: Path,
+    allow_partial: bool = False,
+    item_ids: set[str] | None = None,
+) -> dict[str, Any]:
+    records_dir = output_root / "records"
+    if output_root.exists() and any(output_root.iterdir()):
+        raise FileExistsError(f"refusing to overwrite non-empty {output_root}")
+    records_dir.mkdir(parents=True, exist_ok=True)
+
+    source = load_dataset(source_dataset)
+    if item_ids is not None:
+        unknown = item_ids - set(source)
+        if unknown:
+            raise ValueError(f"unknown requested item IDs: {sorted(unknown)}")
+        source = {
+            item_id: record
+            for item_id, record in source.items()
+            if item_id in item_ids
+        }
+    exported: list[dict[str, Any]] = []
+    missing: list[dict[str, str]] = []
+    for item_id, record in sorted(source.items()):
+        try:
+            variants = {
+                family: read_stage(
+                    surface_run_root,
+                    item_id,
+                    f"surface_{family}_variant",
+                )
+                for family in SURFACE_FAMILIES
+            }
+            kernel_payload, kernel_path = read_kernel(kernel_run_root, item_id)
+        except (FileNotFoundError, ValueError) as exc:
+            missing.append({"item_id": item_id, "reason": str(exc)})
+            continue
+
+        candidate = kernel_payload["accepted_candidate"]
+        variants["kernel_variant"] = {
+            "question": candidate["problem"],
+            "solution": candidate["proof"],
+        }
+        output_record = copy.deepcopy(record)
+        output_record["variants"] = variants
+        output_path = records_dir / f"{safe_name(item_id)}.json"
+        output_path.write_text(
+            json.dumps(output_record, ensure_ascii=False, indent=2) + "\n",
+            encoding="utf-8",
+        )
+        exported.append(
+            {
+                "item_id": item_id,
+                "path": str(output_path.resolve()),
+                "sha256": sha256_file(output_path),
+            }
+        )
+
+    if missing and not allow_partial:
+        # Keep the manifest for diagnosis, but never leave a partial directory
+        # that looks like a completed release.
+        status = "incomplete"
+    else:
+        status = "complete" if not missing else "partial_allowed"
+    manifest = {
+        "status": status,
+        "source_dataset": str(source_dataset.resolve()),
+        "surface_run_root": str(surface_run_root.resolve()),
+        "kernel_run_root": str(kernel_run_root.resolve()),
+        "source_item_count": len(source),
+        "exported_item_count": len(exported),
+        "missing_item_count": len(missing),
+        "allow_partial": allow_partial,
+        "exported": exported,
+        "missing": missing,
+    }
+    (output_root / "manifest.json").write_text(
+        json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    if missing and not allow_partial:
+        raise RuntimeError(
+            f"release incomplete: exported {len(exported)}/{len(source)}; "
+            f"see {output_root / 'manifest.json'}"
+        )
+    return manifest
diff --git a/src/gap_pipeline/store.py b/src/gap_pipeline/store.py
new file mode 100644
index 0000000..ff850b9
--- /dev/null
+++ b/src/gap_pipeline/store.py
@@ -0,0 +1,113 @@
+"""Immutable, hash-addressed JSON artifacts for GAP runs."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel
+
+from .models import ModelCallRecord, StageArtifact
+
+
+def canonical_json(value: Any) -> str:
+    if isinstance(value, BaseModel):
+        value = value.model_dump(mode="json")
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+
+
+def sha256_payload(value: Any) -> str:
+    return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
+
+
+def safe_name(value: str) -> str:
+    return re.sub(r"[^A-Za-z0-9_.-]+", "_", value).strip("._") or "artifact"
+
+
+class RunStore:
+    """Writes immutable artifacts and refuses conflicting overwrites."""
+
+    def __init__(self, root: Path, item_id: str) -> None:
+        self.root = root
+        self.item_id = item_id
+        self.item_root = root / "items" / safe_name(item_id)
+        for directory in [
+            self.item_root,
+            self.item_root / "calls",
+            self.item_root / "stages",
+            self.item_root / "iterations",
+        ]:
+            directory.mkdir(parents=True, exist_ok=True)
+
+    def write_json(self, relative_path: str | Path, value: Any) -> Path:
+        path = self.item_root / relative_path
+        path.parent.mkdir(parents=True, exist_ok=True)
+        payload = value.model_dump(mode="json") if isinstance(value, BaseModel) else value
+        text = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
+        if path.exists():
+            existing = path.read_text(encoding="utf-8")
+            if existing != text:
+                raise FileExistsError(
+                    f"refusing to overwrite non-identical artifact {path}"
+                )
+            return path
+        descriptor, temp_name = tempfile.mkstemp(
+            prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
+        )
+        try:
+            with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
+                handle.write(text)
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(temp_name, path)
+        finally:
+            if os.path.exists(temp_name):
+                os.unlink(temp_name)
+        return path
+
+    def write_input(self, value: Any) -> Path:
+        return self.write_json("input.json", value)
+
+    def write_config(self, value: Any) -> Path:
+        return self.write_json("config.json", value)
+
+    def write_call(self, record: ModelCallRecord) -> Path:
+        return self.write_json(
+            Path("calls") / f"{safe_name(record.request_id)}.json", record
+        )
+
+    def write_stage(
+        self,
+        stage: str,
+        payload: BaseModel | dict[str, Any],
+        *,
+        request_id: str | None,
+    ) -> Path:
+        raw = payload.model_dump(mode="json") if isinstance(payload, BaseModel) else payload
+        artifact = StageArtifact(
+            item_id=self.item_id,
+            stage=stage,
+            payload_sha256=sha256_payload(raw),
+            payload=raw,
+            request_id=request_id,
+        )
+        return self.write_json(Path("stages") / f"{safe_name(stage)}.json", artifact)
+
+    def write_iteration(self, iteration: int, value: Any) -> Path:
+        return self.write_json(
+            Path("iterations") / f"{iteration:02d}.json",
+            value,
+        )
+
+    def write_final(self, value: Any) -> Path:
+        return self.write_json("final.json", value)
diff --git a/src/gap_pipeline/surface.py b/src/gap_pipeline/surface.py
new file mode 100644
index 0000000..7de4e90
--- /dev/null
+++ b/src/gap_pipeline/surface.py
@@ -0,0 +1,217 @@
+"""Surface-renaming generation, application, and deterministic validation."""
+
+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 .store import RunStore
+
+
+SurfaceFamily = Literal[
+    "descriptive_long",
+    "descriptive_long_confusing",
+    "descriptive_long_misleading",
+    "garbled_string",
+]
+
+SURFACE_FAMILIES: tuple[SurfaceFamily, ...] = (
+    "descriptive_long",
+    "descriptive_long_confusing",
+    "descriptive_long_misleading",
+    "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,
+)
+
+
+@dataclass(frozen=True)
+class SurfaceVariant:
+    family: SurfaceFamily
+    rename_map: dict[str, str]
+    problem: str
+    solution: str
+
+    def as_release_payload(self) -> dict[str, object]:
+        return {
+            "map": dict(self.rename_map),
+            "question": self.problem,
+            "solution": self.solution,
+        }
+
+
+def validate_rename_map(
+    rename_map: dict[str, str],
+    *,
+    existing_identifiers: list[str],
+    scientific_constants: list[str],
+) -> None:
+    if not rename_map:
+        raise ValueError("surface rename map cannot be empty")
+    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),
+    )
+
+
+class SurfacePipeline:
+    def __init__(self, proposer: JsonLLM, store: RunStore) -> None:
+        self.proposer = proposer
+        self.store = store
+
+    async def propose_map(
+        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
+            )
+        )
+
+        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,
+                request_id=request_id,
+            )
+            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,
+        )
+        self.store.write_stage(
+            f"surface_{family}_map",
+            {"rename_map": rename_map},
+            request_id=None,
+        )
+        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,
+        )
+        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.
+        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)]