summaryrefslogtreecommitdiff
path: root/src/gap_pipeline
diff options
context:
space:
mode:
Diffstat (limited to 'src/gap_pipeline')
-rw-r--r--src/gap_pipeline/__init__.py24
-rw-r--r--src/gap_pipeline/cli.py168
-rw-r--r--src/gap_pipeline/clients.py168
-rw-r--r--src/gap_pipeline/e2e.py301
-rw-r--r--src/gap_pipeline/models.py294
-rw-r--r--src/gap_pipeline/offline.py279
-rw-r--r--src/gap_pipeline/pipeline.py459
-rw-r--r--src/gap_pipeline/prompts.py338
-rw-r--r--src/gap_pipeline/release.py138
-rw-r--r--src/gap_pipeline/store.py113
-rw-r--r--src/gap_pipeline/surface.py217
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)]
generated by cgit v1.2.3 (git 2.25.1) at 2026-07-24 23:06:50 +0000