diff options
| -rw-r--r-- | .gitignore | 12 | ||||
| -rw-r--r-- | GAP_End_to_End.ipynb | 169 | ||||
| -rw-r--r-- | README.md | 105 | ||||
| -rw-r--r-- | STAGE_MAP.md | 35 | ||||
| -rw-r--r-- | examples/sample_data/1998-B-1.json | 12 | ||||
| -rw-r--r-- | pyproject.toml | 26 | ||||
| -rw-r--r-- | src/gap_pipeline/__init__.py | 24 | ||||
| -rw-r--r-- | src/gap_pipeline/cli.py | 168 | ||||
| -rw-r--r-- | src/gap_pipeline/clients.py | 168 | ||||
| -rw-r--r-- | src/gap_pipeline/e2e.py | 301 | ||||
| -rw-r--r-- | src/gap_pipeline/models.py | 294 | ||||
| -rw-r--r-- | src/gap_pipeline/offline.py | 279 | ||||
| -rw-r--r-- | src/gap_pipeline/pipeline.py | 459 | ||||
| -rw-r--r-- | src/gap_pipeline/prompts.py | 338 | ||||
| -rw-r--r-- | src/gap_pipeline/release.py | 138 | ||||
| -rw-r--r-- | src/gap_pipeline/store.py | 113 | ||||
| -rw-r--r-- | src/gap_pipeline/surface.py | 217 | ||||
| -rw-r--r-- | tests/conftest.py | 126 | ||||
| -rw-r--r-- | tests/test_e2e.py | 30 | ||||
| -rw-r--r-- | tests/test_models.py | 71 | ||||
| -rw-r--r-- | tests/test_offline.py | 6 | ||||
| -rw-r--r-- | tests/test_pipeline.py | 236 | ||||
| -rw-r--r-- | tests/test_release.py | 80 | ||||
| -rw-r--r-- | tests/test_store.py | 13 | ||||
| -rw-r--r-- | tests/test_surface.py | 43 |
25 files changed, 3463 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..51b0cb6 --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +.venv/ +build/ +dist/ +*.egg-info/ +runs/ +surface-runs/ +kernel-runs/ +release/ +notebook_runs/ diff --git a/GAP_End_to_End.ipynb b/GAP_End_to_End.ipynb new file mode 100644 index 0000000..92910bb --- /dev/null +++ b/GAP_End_to_End.ipynb @@ -0,0 +1,169 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "gap-title", + "metadata": {}, + "source": [ + "# GAP end-to-end reproduction\n", + "\n", + "Run all cells to generate four surface variants and one verified kernel variant for a Putnam problem, then export a machine-readable GAP record.\n", + "\n", + "The live path uses `o3` for generation and five independent verification calls per round. At least two unanimous rounds are required, so a live run makes multiple API calls and may take several minutes. The API key is read with `getpass` and is never written into the notebook or run artifacts." + ] + }, + { + "cell_type": "code", + "id": "install-package", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from pathlib import Path\n", + "import json\n", + "import os\n", + "import subprocess\n", + "import sys\n", + "from datetime import datetime, timezone\n", + "\n", + "candidates = [Path.cwd(), Path.cwd().parent]\n", + "PACKAGE_ROOT = next((p.resolve() for p in candidates if (p / \"pyproject.toml\").exists()), None)\n", + "if PACKAGE_ROOT is None:\n", + " raise RuntimeError(\"Open this notebook from the GAP package root or its parent directory.\")\n", + "\n", + "subprocess.run(\n", + " [sys.executable, \"-m\", \"pip\", \"install\", \"-q\", \"-e\", f\"{PACKAGE_ROOT}[api,test]\"],\n", + " check=True,\n", + ")\n", + "src_path = str(PACKAGE_ROOT / \"src\")\n", + "if src_path not in sys.path:\n", + " sys.path.insert(0, src_path)\n", + "print(f\"GAP package root: {PACKAGE_ROOT}\")" + ] + }, + { + "cell_type": "code", + "id": "configure-run", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Configuration. Set GAP_OFFLINE_SMOKE=1 to exercise the full pipeline without an API call.\n", + "OFFLINE_SMOKE = os.environ.get(\"GAP_OFFLINE_SMOKE\", \"0\") == \"1\"\n", + "MODEL = os.environ.get(\"GAP_MODEL\", \"o3\")\n", + "ITEM_ID = os.environ.get(\"GAP_ITEM_ID\", \"1998-B-1\")\n", + "\n", + "dataset_candidates = [\n", + " Path(os.environ[\"PUTNAMGAP_DATASET\"]).expanduser() if os.environ.get(\"PUTNAMGAP_DATASET\") else None,\n", + " PACKAGE_ROOT / \"examples\" / \"sample_data\",\n", + " PACKAGE_ROOT.parent / \"putnamsup\" / \"PutnamGAP\",\n", + "]\n", + "DATASET_DIR = next((p.resolve() for p in dataset_candidates if p is not None and p.exists()), None)\n", + "if not OFFLINE_SMOKE and DATASET_DIR is None:\n", + " raise RuntimeError(\"Set PUTNAMGAP_DATASET to the directory containing the PutnamGAP JSON files.\")\n", + "\n", + "stamp = datetime.now(timezone.utc).strftime(\"%Y%m%dT%H%M%SZ\")\n", + "OUTPUT_BASE = Path(os.environ.get(\"GAP_NOTEBOOK_OUTPUT_BASE\", PACKAGE_ROOT / \"notebook_runs\")).expanduser()\n", + "WORK_ROOT = OUTPUT_BASE / f\"{ITEM_ID}-{stamp}\"\n", + "print({\"offline_smoke\": OFFLINE_SMOKE, \"model\": MODEL, \"item_id\": ITEM_ID, \"dataset\": str(DATASET_DIR), \"work_root\": str(WORK_ROOT)})" + ] + }, + { + "cell_type": "code", + "id": "load-api-key", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "api_key = None\n", + "if not OFFLINE_SMOKE:\n", + " api_key = os.environ.get(\"OPENAI_API_KEY\")\n", + " if not api_key:\n", + " from getpass import getpass\n", + " api_key = getpass(\"OPENAI_API_KEY (input hidden): \").strip()\n", + " if not api_key:\n", + " raise RuntimeError(\"An OpenAI API key is required for the live run.\")\n", + " print(\"API key loaded in memory; it will not be displayed or saved.\")\n", + "else:\n", + " print(\"Offline scripted smoke mode: no API key required.\")" + ] + }, + { + "cell_type": "code", + "id": "run-pipeline", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from gap_pipeline.e2e import run_live_item, run_offline_smoke\n", + "\n", + "if OFFLINE_SMOKE:\n", + " result = await run_offline_smoke(WORK_ROOT)\n", + "else:\n", + " result = await run_live_item(\n", + " dataset_dir=DATASET_DIR,\n", + " item_id=ITEM_ID,\n", + " work_root=WORK_ROOT,\n", + " model=MODEL,\n", + " api_key=api_key,\n", + " )\n", + "\n", + "print(json.dumps(result, indent=2, ensure_ascii=False))" + ] + }, + { + "cell_type": "code", + "id": "verify-output", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# End-to-end assertions and a compact preview of the exported record.\n", + "record = json.loads(Path(result[\"release_record\"]).read_text(encoding=\"utf-8\"))\n", + "expected_families = {\n", + " \"descriptive_long\",\n", + " \"descriptive_long_confusing\",\n", + " \"descriptive_long_misleading\",\n", + " \"garbled_string\",\n", + " \"kernel_variant\",\n", + "}\n", + "assert set(record[\"variants\"]) == expected_families\n", + "assert result[\"kernel_status\"] == \"accepted\"\n", + "assert result[\"verification_rounds\"] >= 2\n", + "assert result[\"export_status\"] == \"complete\"\n", + "\n", + "kernel = record[\"variants\"][\"kernel_variant\"]\n", + "print(\"END-TO-END PASS\")\n", + "print(f\"Item: {record['index']}\")\n", + "print(f\"Variant families: {sorted(record['variants'])}\")\n", + "print(f\"Verification rounds: {result['verification_rounds']}\")\n", + "print(f\"Kernel question preview: {kernel['question'][:500]}\")\n", + "print(f\"Exported record: {result['release_record']}\")\n", + "print(f\"Manifest: {result['manifest']}\")" + ] + }, + { + "cell_type": "markdown", + "id": "full-dataset-note", + "metadata": {}, + "source": [ + "## Run another released item\n", + "\n", + "The notebook intentionally runs one item so the complete path can be checked at modest cost. Set `PUTNAMGAP_DATASET` and `GAP_ITEM_ID` before running to choose another source item; repeat the same runner over IDs for a larger generation job." + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.11" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/README.md b/README.md new file mode 100644 index 0000000..56b76ec --- /dev/null +++ b/README.md @@ -0,0 +1,105 @@ +# GAP minimal reproduction package + +This repository contains the generation and verification pipeline for GAP: +four surface-renaming families and one proof-plan-preserving kernel variant. +It is intentionally small and includes one canonical Putnam example for an +end-to-end check. + +## One-click reproduction + +Open `GAP_End_to_End.ipynb` and choose **Run All**. The notebook: + +1. installs the package; +2. loads one canonical problem; +3. generates four surface variants; +4. runs the five-stage kernel pipeline; +5. applies five judges until the same candidate receives two consecutive + unanimous rounds; +6. exports and validates one machine-readable GAP record. + +The default live model is `o3`. Set `OPENAI_API_KEY` before starting Jupyter, +or enter it in the notebook's hidden prompt. The key is not written to the +notebook or run artifacts. + +To test the complete software path without API calls: + +```bash +GAP_OFFLINE_SMOKE=1 jupyter notebook GAP_End_to_End.ipynb +``` + +Offline smoke mode uses scripted model responses. It validates pipeline wiring, +verification control flow, and release assembly; it does not evaluate generated +mathematical content. + +## Install and test + +```bash +python -m pip install -e '.[api,test]' +pytest +``` + +Without installation: + +```bash +PYTHONPATH=src pytest +PYTHONPATH=src python -m gap_pipeline.cli --help +``` + +## Live one-item commands + +Generate the kernel variant: + +```bash +export OPENAI_API_KEY=... +PYTHONPATH=src python -m gap_pipeline.cli generate-kernel \ + --dataset examples/sample_data \ + --item-id 1998-B-1 \ + --run-dir runs/kernel \ + --proposer-model o3 \ + --judge-model o3 +``` + +Generate all surface variants: + +```bash +PYTHONPATH=src python -m gap_pipeline.cli generate-surfaces \ + --dataset examples/sample_data \ + --item-id 1998-B-1 \ + --run-dir runs/surface \ + --proposer-model o3 +``` + +Assemble the final record after both runs complete: + +```bash +PYTHONPATH=src python -m gap_pipeline.cli export-release \ + --source-dataset examples/sample_data \ + --surface-run-dir runs/surface \ + --kernel-run-dir runs/kernel \ + --output-root release \ + --item-id 1998-B-1 +``` + +## Protocol + +The kernel path consists of: + +1. reference solution → proof DAG; +2. proof DAG → content-free method plan; +3. guarded replacement at a DAG leaf; +4. replacement diffusion through the fixed DAG; +5. regenerated proof → self-contained problem statement. + +Verification uses `J=5` judges, requires `K=2` consecutive unanimous rounds for +the same candidate, and permits at most `T=15` rounds. A non-unanimous round +resets the pass streak and triggers repair. + +`STAGE_MAP.md` maps each paper operation to the implementation and its validated +output. Every model call and intermediate stage is saved below the selected run +directory for inspection. + +## Reproducibility scope + +LLM generation is stochastic, so repeated live runs need not produce identical +wording. This package reproduces the prompts, transformation stages, typed +structural checks, verification state machine, and release format. diff --git a/STAGE_MAP.md b/STAGE_MAP.md new file mode 100644 index 0000000..1cf0691 --- /dev/null +++ b/STAGE_MAP.md @@ -0,0 +1,35 @@ +# GAP paper-to-code map + +| Paper operation | Implementation | Validated output | +|---|---|---| +| Surface rename | `surface.py` | `SurfaceVariant` with a collision-free one-to-one identifier map | +| 1. Reference solution to proof DAG | `KernelPipeline.construct_dag` | Acyclic `ProofDAG` connected to the terminal claim | +| 2. Content-free method plan | `KernelPipeline.summarize_methods` | `MethodPlan` with the DAG's node IDs and topological order | +| 3. Guarded leaf replacement | `KernelPipeline.generate_replacement` | `ReplacementSpec` with explicit guards and evidence | +| 4. Diffusion through the DAG | `KernelPipeline.diffuse_dag` | `DiffusedProof` preserving node order, dependencies, and method labels | +| 5. Problem rendering | `KernelPipeline.render_question` | Self-contained `KernelCandidate` aligned with the regenerated proof | +| Five-judge verification | `KernelPipeline.verify_candidate` | Five `JudgeVerdict` objects per round | +| Consecutive-pass protocol | `KernelPipeline.verify_candidate` | `K=2`; rejection resets the streak | +| Repair loop | `KernelPipeline._repair` | Corrected candidate with a distinct content hash; maximum `T=15` rounds | +| Audit sampling flag | `KernelPipeline._audit_selected` | Deterministic 10% post-hoc audit selection | + +## Per-item artifacts + +```text +items/<item-id>/ + input.json + config.json + calls/<request-id>.json + stages/<stage>.json + iterations/<round>.json + final.json +``` + +Each JSON artifact includes a SHA-256 digest of its canonical payload. Judge +calls remain separate rather than being collapsed into a vote count. + +## Claim boundary + +The verifier is a filtering protocol, not a formal proof of mathematical +correctness. Accepted variants should additionally be inspected in the +post-hoc human audit described in the accompanying paper. diff --git a/examples/sample_data/1998-B-1.json b/examples/sample_data/1998-B-1.json new file mode 100644 index 0000000..4e9aa4b --- /dev/null +++ b/examples/sample_data/1998-B-1.json @@ -0,0 +1,12 @@ +{ + "index": "1998-B-1", + "question": "Find the minimum value of\n\\[\n\\frac{(x+1/x)^6-(x^6+1/x^6)-2}{(x+1/x)^3+(x^3+1/x^3)}\n\\]\nfor $x>0$.", + "solution": "Notice that\n\\[\n\\frac{(x+1/x)^6-(x^6+1/x^6)-2}{(x+1/x)^3+(x^3+1/x^3)}\n=(x+1/x)^3-(x^3+1/x^3)=3(x+1/x).\n\\]\nThe last expression has minimum value $6$ by AM-GM, achieved at $x=1$.", + "vars": [ + "x" + ], + "params": [], + "sci_consts": [], + "problem_type": "calculation", + "variants": {} +} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ce2afe9 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,26 @@ +[build-system] +requires = ["setuptools>=69"] +build-backend = "setuptools.build_meta" + +[project] +name = "gap-reproduce" +version = "0.1.0" +description = "Minimal reproduction package for the GAP mathematical transformation pipeline" +requires-python = ">=3.11" +dependencies = [ + "pydantic>=2.8", +] + +[project.optional-dependencies] +api = ["openai>=1.68"] +test = ["pytest>=8"] + +[project.scripts] +gap-reproduce = "gap_pipeline.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-q" 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)] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..80ccc2e --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +import copy + +import pytest + +from gap_pipeline.models import CanonicalItem + + +@pytest.fixture +def item() -> CanonicalItem: + return CanonicalItem( + item_id="demo-A-1", + problem=r"Let \(a>0\). Prove that \(a+1/a\ge 2\).", + solution=( + r"Since \((a-1)^2\ge0\), expanding gives \(a^2-2a+1\ge0\). " + r"Divide by \(a>0\) to obtain \(a+1/a\ge2\)." + ), + variables=["a"], + problem_type="proof", + ) + + +@pytest.fixture +def dag_dict() -> dict: + return { + "nodes": [ + { + "node_id": "n1", + "claim": "(a-1)^2 >= 0", + "derivation": "a square is nonnegative", + "dependencies": [], + "source_span": "Since (a-1)^2 >= 0", + "mathematical_objects": ["a", "(a-1)^2"], + }, + { + "node_id": "n2", + "claim": "a+1/a >= 2", + "derivation": "expand and divide by positive a", + "dependencies": ["n1"], + "source_span": "Divide by a>0", + "mathematical_objects": ["a"], + }, + ], + "terminal_node_id": "n2", + } + + +@pytest.fixture +def plan_dict() -> dict: + return { + "steps": [ + { + "node_id": "n1", + "method_label": "use nonnegativity of a square", + "input_roles": ["positive scalar"], + "output_role": "nonnegative expression", + "invariants": ["scalar is real"], + }, + { + "node_id": "n2", + "method_label": "expand and divide by a positive quantity", + "input_roles": ["nonnegative expression", "positive scalar"], + "output_role": "target inequality", + "invariants": ["divisor is positive"], + }, + ], + "terminal_node_id": "n2", + } + + +@pytest.fixture +def replacement_dict() -> dict: + return { + "target_node_id": "n1", + "original_object": "(a-1)^2", + "replacement_object": "(x-2)^2", + "guard_conditions": ["x>0"], + "guard_evidence": ["choose x in the positive reals"], + "expected_downstream_changes": ["expand around 2", "divide by x"], + "rationale": "the same square-nonnegativity plan applies", + } + + +@pytest.fixture +def diffused_dict(plan_dict: dict) -> dict: + return { + "node_instantiations": [ + { + "node_id": "n1", + "method_label": plan_dict["steps"][0]["method_label"], + "instantiated_claim": "(x-2)^2 >= 0", + "instantiated_derivation": "a square is nonnegative", + "dependencies": [], + }, + { + "node_id": "n2", + "method_label": plan_dict["steps"][1]["method_label"], + "instantiated_claim": "x+4/x >= 4", + "instantiated_derivation": "expand and divide by positive x", + "dependencies": ["n1"], + }, + ], + "regenerated_proof": ( + "Since (x-2)^2 >= 0, expand and divide by x>0 to get x+4/x >= 4." + ), + "terminal_answer": "x+4/x >= 4", + } + + +@pytest.fixture +def candidate_dict(replacement_dict: dict, diffused_dict: dict) -> dict: + return { + "problem": "Let x>0. Prove that x+4/x >= 4.", + "proof": diffused_dict["regenerated_proof"], + "terminal_answer": diffused_dict["terminal_answer"], + "node_instantiations": diffused_dict["node_instantiations"], + "replacement": replacement_dict, + } + + +def changed_candidate(candidate: dict, suffix: str) -> dict: + value = copy.deepcopy(candidate) + value["problem"] = value["problem"] + f" [{suffix}]" + value["proof"] = value["proof"] + f" Repair {suffix}." + return value diff --git a/tests/test_e2e.py b/tests/test_e2e.py new file mode 100644 index 0000000..9da1190 --- /dev/null +++ b/tests/test_e2e.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import asyncio +import json +from pathlib import Path + +from gap_pipeline.e2e import run_offline_smoke + + +def test_offline_end_to_end_smoke(tmp_path: Path) -> None: + result = asyncio.run(run_offline_smoke(tmp_path / "e2e")) + assert result["mode"] == "offline-smoke" + assert result["surface_families"] == [ + "descriptive_long", + "descriptive_long_confusing", + "descriptive_long_misleading", + "garbled_string", + ] + assert result["kernel_status"] == "accepted" + assert result["verification_rounds"] == 2 + assert result["export_status"] == "complete" + assert result["exported_item_count"] == 1 + record = json.loads(Path(result["release_record"]).read_text()) + assert set(record["variants"]) == { + "descriptive_long", + "descriptive_long_confusing", + "descriptive_long_misleading", + "garbled_string", + "kernel_variant", + } diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..2c678a0 --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from gap_pipeline.models import ( + DiffusedProof, + MethodPlan, + ProofDAG, + ReplacementSpec, +) + + +def test_valid_dag_and_plan(dag_dict: dict, plan_dict: dict) -> None: + dag = ProofDAG.model_validate(dag_dict) + assert [node.node_id for node in dag.topological_nodes()] == ["n1", "n2"] + assert dag.leaf_node_ids() == ["n1"] + plan = MethodPlan.model_validate(plan_dict) + plan.validate_against(dag) + + +def test_cycle_is_rejected(dag_dict: dict) -> None: + dag_dict["nodes"][0]["dependencies"] = ["n2"] + with pytest.raises(ValidationError, match="cycle"): + ProofDAG.model_validate(dag_dict) + + +def test_node_disconnected_from_terminal_is_rejected(dag_dict: dict) -> None: + dag_dict["nodes"].append( + { + "node_id": "orphan", + "claim": "unused claim", + "derivation": "unused derivation", + "dependencies": [], + "source_span": "", + "mathematical_objects": [], + } + ) + with pytest.raises(ValidationError, match="disconnected"): + ProofDAG.model_validate(dag_dict) + + +def test_method_reordering_is_rejected(dag_dict: dict, plan_dict: dict) -> None: + dag = ProofDAG.model_validate(dag_dict) + plan_dict["steps"].reverse() + plan = MethodPlan.model_validate(plan_dict) + with pytest.raises(ValueError, match="topological order"): + plan.validate_against(dag) + + +def test_replacement_must_target_leaf( + dag_dict: dict, replacement_dict: dict +) -> None: + dag = ProofDAG.model_validate(dag_dict) + replacement_dict["target_node_id"] = "n2" + replacement = ReplacementSpec.model_validate(replacement_dict) + with pytest.raises(ValueError, match="not a DAG leaf"): + replacement.validate_against(dag) + + +def test_diffusion_cannot_change_dag_dependencies( + dag_dict: dict, + plan_dict: dict, + diffused_dict: dict, +) -> None: + dag = ProofDAG.model_validate(dag_dict) + plan = MethodPlan.model_validate(plan_dict) + diffused_dict["node_instantiations"][1]["dependencies"] = [] + diffused = DiffusedProof.model_validate(diffused_dict) + with pytest.raises(ValueError, match="changes dependencies"): + diffused.validate_against(dag, plan) diff --git a/tests/test_offline.py b/tests/test_offline.py new file mode 100644 index 0000000..04b7f46 --- /dev/null +++ b/tests/test_offline.py @@ -0,0 +1,6 @@ +from gap_pipeline.offline import normalize_latex_symbol + + +def test_latex_symbol_aliases_normalize_together() -> None: + assert normalize_latex_symbol(r"x_{n}") == normalize_latex_symbol("x_n") + assert normalize_latex_symbol(r"\\phi") == normalize_latex_symbol(r"\phi") diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..ea1077a --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import asyncio +import json + +from gap_pipeline.clients import ScriptedClient +from gap_pipeline.pipeline import KernelPipeline, PipelineConfig +from gap_pipeline.store import RunStore + +from conftest import changed_candidate + + +def accept_verdict() -> dict: + return { + "verdict": "accept", + "step_by_step_check": "n1 passes; n2 passes", + "blocking_issues": "", + "patch_suggestion": "", + } + + +def reject_verdict(iteration: int) -> dict: + return { + "verdict": "reject", + "step_by_step_check": f"n1 passes; n2 fails at iteration {iteration}", + "blocking_issues": "the terminal algebra needs correction", + "patch_suggestion": "correct the terminal algebra without changing the plan", + } + + +def proposer_responses( + item_id: str, + dag_dict: dict, + plan_dict: dict, + replacement_dict: dict, + diffused_dict: dict, + candidate_dict: dict, +) -> dict: + return { + f"{item_id}.stage1": dag_dict, + f"{item_id}.stage2": plan_dict, + f"{item_id}.stage3": replacement_dict, + f"{item_id}.stage4": diffused_dict, + f"{item_id}.stage5": candidate_dict, + } + + +def test_two_consecutive_unanimous_passes_accept( + tmp_path, + item, + dag_dict, + plan_dict, + replacement_dict, + diffused_dict, + candidate_dict, +) -> None: + proposer = ScriptedClient( + proposer_responses( + item.item_id, + dag_dict, + plan_dict, + replacement_dict, + diffused_dict, + candidate_dict, + ) + ) + judges = [ + ScriptedClient({f"{item.item_id}.verify": [accept_verdict(), accept_verdict()]}) + for _ in range(5) + ] + pipeline = KernelPipeline( + proposer=proposer, + judges=judges, + store=RunStore(tmp_path, item.item_id), + config=PipelineConfig(proposer_model="scripted", judge_model="scripted"), + ) + result = asyncio.run(pipeline.run(item)) + assert result.status == "accepted" + assert len(result.iterations) == 2 + assert [row.pass_streak_after for row in result.iterations] == [1, 2] + assert not any(row.repaired for row in result.iterations) + + final_path = tmp_path / "items" / item.item_id / "final.json" + final = json.loads(final_path.read_text()) + assert final["human_audit_selected"] in {True, False} + assert len(list((final_path.parent / "calls").glob("*.json"))) == 15 + + +def test_rejection_resets_streak_and_repairs( + tmp_path, + item, + dag_dict, + plan_dict, + replacement_dict, + diffused_dict, + candidate_dict, +) -> None: + repaired = changed_candidate(candidate_dict, "fixed") + responses = proposer_responses( + item.item_id, + dag_dict, + plan_dict, + replacement_dict, + diffused_dict, + candidate_dict, + ) + responses[f"{item.item_id}.repair"] = repaired + proposer = ScriptedClient(responses) + judges = [] + for judge_id in range(1, 6): + first = reject_verdict(1) if judge_id == 1 else accept_verdict() + judges.append( + ScriptedClient( + { + f"{item.item_id}.verify": [ + first, + accept_verdict(), + accept_verdict(), + ] + } + ) + ) + pipeline = KernelPipeline( + proposer=proposer, + judges=judges, + store=RunStore(tmp_path, item.item_id), + config=PipelineConfig(proposer_model="scripted", judge_model="scripted"), + ) + result = asyncio.run(pipeline.run(item)) + assert result.status == "accepted" + assert len(result.iterations) == 3 + assert result.iterations[0].repaired + assert [row.pass_streak_after for row in result.iterations] == [0, 1, 2] + assert ( + result.iterations[0].repaired_candidate_sha256 + == result.iterations[1].candidate_sha256 + == result.iterations[2].candidate_sha256 + ) + + +def test_rejection_after_one_pass_resets_streak_on_new_candidate( + tmp_path, + item, + dag_dict, + plan_dict, + replacement_dict, + diffused_dict, + candidate_dict, +) -> None: + repaired = changed_candidate(candidate_dict, "after-broken-streak") + responses = proposer_responses( + item.item_id, + dag_dict, + plan_dict, + replacement_dict, + diffused_dict, + candidate_dict, + ) + responses[f"{item.item_id}.repair"] = repaired + proposer = ScriptedClient(responses) + judges = [] + for judge_id in range(1, 6): + second = reject_verdict(2) if judge_id == 3 else accept_verdict() + judges.append( + ScriptedClient( + { + f"{item.item_id}.verify": [ + accept_verdict(), + second, + accept_verdict(), + accept_verdict(), + ] + } + ) + ) + pipeline = KernelPipeline( + proposer=proposer, + judges=judges, + store=RunStore(tmp_path, item.item_id), + config=PipelineConfig(proposer_model="scripted", judge_model="scripted"), + ) + result = asyncio.run(pipeline.run(item)) + assert result.status == "accepted" + assert [row.pass_streak_after for row in result.iterations] == [1, 0, 1, 2] + assert result.iterations[0].candidate_sha256 == result.iterations[1].candidate_sha256 + assert result.iterations[1].repaired_candidate_sha256 != result.iterations[1].candidate_sha256 + assert ( + result.iterations[1].repaired_candidate_sha256 + == result.iterations[2].candidate_sha256 + == result.iterations[3].candidate_sha256 + ) + + +def test_fifteen_failed_rounds_reject( + tmp_path, + item, + dag_dict, + plan_dict, + replacement_dict, + diffused_dict, + candidate_dict, +) -> None: + responses = proposer_responses( + item.item_id, + dag_dict, + plan_dict, + replacement_dict, + diffused_dict, + candidate_dict, + ) + responses[f"{item.item_id}.repair"] = [ + changed_candidate(candidate_dict, f"repair-{iteration}") + for iteration in range(1, 15) + ] + proposer = ScriptedClient(responses) + judges = [ + ScriptedClient( + { + f"{item.item_id}.verify": [ + reject_verdict(iteration) for iteration in range(1, 16) + ] + } + ) + for _ in range(5) + ] + pipeline = KernelPipeline( + proposer=proposer, + judges=judges, + store=RunStore(tmp_path, item.item_id), + config=PipelineConfig(proposer_model="scripted", judge_model="scripted"), + ) + result = asyncio.run(pipeline.run(item)) + assert result.status == "rejected" + assert len(result.iterations) == 15 + assert sum(row.repaired for row in result.iterations) == 14 + assert all(row.pass_streak_after == 0 for row in result.iterations) diff --git a/tests/test_release.py b/tests/test_release.py new file mode 100644 index 0000000..3653ff1 --- /dev/null +++ b/tests/test_release.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import json + +import pytest + +from gap_pipeline.models import KernelCandidate +from gap_pipeline.release import export_release +from gap_pipeline.store import RunStore, sha256_payload +from gap_pipeline.surface import SURFACE_FAMILIES + + +def test_offline_release_export_verifies_and_assembles( + tmp_path, + item, + candidate_dict, +) -> None: + source_dir = tmp_path / "source" + source_dir.mkdir() + source_record = { + "index": item.item_id, + "question": item.problem, + "solution": item.solution, + "vars": item.variables, + "params": [], + "sci_consts": [], + "problem_type": "proof", + "variants": {}, + } + (source_dir / f"{item.item_id}.json").write_text( + json.dumps(source_record), + encoding="utf-8", + ) + + surface_root = tmp_path / "surface-runs" + surface_store = RunStore(surface_root, item.item_id) + for family in SURFACE_FAMILIES: + surface_store.write_stage( + f"surface_{family}_variant", + { + "map": {"a": f"name{family.replace('_', '')}"}, + "question": f"{family} question", + "solution": f"{family} solution", + }, + request_id=None, + ) + + kernel_root = tmp_path / "kernel-runs" + kernel_store = RunStore(kernel_root, item.item_id) + candidate = KernelCandidate.model_validate(candidate_dict) + kernel_store.write_final( + { + "item_id": item.item_id, + "status": "accepted", + "accepted_candidate": candidate.model_dump(mode="json"), + "accepted_candidate_sha256": sha256_payload(candidate), + } + ) + + output_root = tmp_path / "release" + manifest = export_release( + source_dataset=source_dir, + surface_run_root=surface_root, + kernel_run_root=kernel_root, + output_root=output_root, + ) + assert manifest["status"] == "complete" + assert manifest["exported_item_count"] == 1 + output = json.loads( + (output_root / "records" / f"{item.item_id}.json").read_text() + ) + assert set(output["variants"]) == {*SURFACE_FAMILIES, "kernel_variant"} + assert output["variants"]["kernel_variant"]["question"] == candidate.problem + with pytest.raises(FileExistsError): + export_release( + source_dataset=source_dir, + surface_run_root=surface_root, + kernel_run_root=kernel_root, + output_root=output_root, + ) diff --git a/tests/test_store.py b/tests/test_store.py new file mode 100644 index 0000000..f944130 --- /dev/null +++ b/tests/test_store.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +import pytest + +from gap_pipeline.store import RunStore + + +def test_store_refuses_conflicting_overwrite(tmp_path) -> None: + store = RunStore(tmp_path, "item") + store.write_json("artifact.json", {"value": 1}) + store.write_json("artifact.json", {"value": 1}) + with pytest.raises(FileExistsError, match="refusing to overwrite"): + store.write_json("artifact.json", {"value": 2}) diff --git a/tests/test_surface.py b/tests/test_surface.py new file mode 100644 index 0000000..78c4d5e --- /dev/null +++ b/tests/test_surface.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import pytest + +from gap_pipeline.surface import ( + apply_rename_map, + create_surface_variant, + validate_rename_map, +) + + +def test_math_only_rename_preserves_english_article(item) -> None: + text = r"Let \(a\) be a randomly chosen value with \(a>0\)." + renamed = apply_rename_map(text, {"a": "coefalpha"}) + assert renamed == ( + r"Let \(coefalpha\) be a randomly chosen value with \(coefalpha>0\)." + ) + + +def test_surface_variant_changes_problem_and_solution(item) -> None: + variant = create_surface_variant( + item, + "descriptive_long", + {"a": "positiveScalar"}, + ) + assert "positiveScalar" in variant.problem + assert "positiveScalar" in variant.solution + assert "Let positiveScalar" not in variant.problem + + +def test_collision_and_nonbijection_are_rejected(item) -> None: + with pytest.raises(ValueError, match="collides"): + validate_rename_map( + {"a": "a"}, + existing_identifiers=["a"], + scientific_constants=[], + ) + with pytest.raises(ValueError, match="one-to-one"): + validate_rename_map( + {"a": "sharedName", "b": "sharedName"}, + existing_identifiers=["a", "b"], + scientific_constants=[], + ) |