From 15efc30e9e7179accd30375d3edb2e34a3b4dc5f Mon Sep 17 00:00:00 2001 From: Oscar Wan Date: Fri, 24 Jul 2026 20:45:42 -0700 Subject: updated generation process --- src/gap_pipeline/cli.py | 6 +- src/gap_pipeline/e2e.py | 104 ++++++++-- src/gap_pipeline/kernel_models.py | 348 ++++++++++++++++++++++++++++++++ src/gap_pipeline/kernel_prompts.py | 276 ++++++++++++++++++++++++++ src/gap_pipeline/models.py | 17 +- src/gap_pipeline/offline.py | 14 +- src/gap_pipeline/paper_pipeline.py | 395 +++++++++++++++++++++++++++++++++++++ src/gap_pipeline/pipeline.py | 5 +- src/gap_pipeline/release.py | 13 ++ src/gap_pipeline/surface.py | 29 ++- 10 files changed, 1175 insertions(+), 32 deletions(-) create mode 100644 src/gap_pipeline/kernel_models.py create mode 100644 src/gap_pipeline/kernel_prompts.py create mode 100644 src/gap_pipeline/paper_pipeline.py (limited to 'src/gap_pipeline') diff --git a/src/gap_pipeline/cli.py b/src/gap_pipeline/cli.py index 105cc97..e79e3e1 100644 --- a/src/gap_pipeline/cli.py +++ b/src/gap_pipeline/cli.py @@ -16,7 +16,7 @@ from .offline import ( summarize_run, validate_public_dataset, ) -from .pipeline import KernelPipeline, PipelineConfig +from .paper_pipeline import PaperKernelPipeline, PaperPipelineConfig from .release import export_release from .store import RunStore from .surface import SurfacePipeline @@ -36,7 +36,7 @@ async def generate_kernel(args: argparse.Namespace) -> None: 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( + config = PaperPipelineConfig( proposer_model=args.proposer_model, judge_model=args.judge_model, ) @@ -48,7 +48,7 @@ async def generate_kernel(args: argparse.Namespace) -> None: ) for judge_id in range(1, 6) ] - pipeline = KernelPipeline( + pipeline = PaperKernelPipeline( proposer=proposer, judges=judges, store=RunStore(args.run_dir, item.item_id), diff --git a/src/gap_pipeline/e2e.py b/src/gap_pipeline/e2e.py index ded1381..66101e7 100644 --- a/src/gap_pipeline/e2e.py +++ b/src/gap_pipeline/e2e.py @@ -11,7 +11,7 @@ from typing import Any from .clients import OpenAIJsonClient, ScriptedClient from .models import CanonicalItem from .offline import load_dataset -from .pipeline import KernelPipeline, PipelineConfig +from .paper_pipeline import PaperKernelPipeline, PaperPipelineConfig from .release import export_release from .store import RunStore from .surface import SurfacePipeline @@ -25,7 +25,8 @@ def _ensure_fresh(path: Path) -> None: def _review_accept() -> dict[str, str]: return { "verdict": "accept", - "step_by_step_check": "n1 passes; n2 passes", + "step_by_step_check": "n1 passes; n2 passes; n3 passes", + "replacement_check": "s1 satisfies its positivity guard", "blocking_issues": "", "patch_suggestion": "", } @@ -84,8 +85,8 @@ async def run_live_item( surface_store, ).run_all(item) - config = PipelineConfig(proposer_model=model, judge_model=model) - kernel = await KernelPipeline( + config = PaperPipelineConfig(proposer_model=model, judge_model=model) + kernel = await PaperKernelPipeline( 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), @@ -164,24 +165,91 @@ async def run_offline_smoke(work_root: Path) -> dict[str, Any]: proposer = ScriptedClient( { - f"{item_id}.plan": { - "core_steps": [ - "use nonnegativity of a square", - "expand and divide by a positive quantity", + f"{item_id}.stage1.dag": { + "nodes": [ + { + "node_id": "n1", + "claim": "(a-1)^2 >= 0", + "dependencies": [], + }, + { + "node_id": "n2", + "claim": "a^2-2a+1 >= 0", + "dependencies": ["n1"], + }, + { + "node_id": "n3", + "claim": "a+1/a >= 2", + "dependencies": ["n2"], + }, ], - "mutable_slots": { - "slot1": { - "description": "the positive reference value", - "original": "1", + "terminal_node_id": "n3", + }, + f"{item_id}.stage2.methods": { + "nodes": [ + { + "node_id": "n1", + "method_label": "use nonnegativity of a square", + }, + { + "node_id": "n2", + "method_label": "expand the square", + }, + { + "node_id": "n3", + "method_label": "divide by a positive quantity", + }, + ] + }, + f"{item_id}.stage3.replacement": { + "changes": [ + { + "slot_id": "s1", + "source_node_id": "n1", + "description": "positive square root at equality", + "original_value": "1", + "replacement_value": "2", + "guard_condition": "replacement is positive", + "guard_justification": "2 is positive", } - }, + ], + "closure_statement": "The equality value is the only change.", + }, + f"{item_id}.stage4.diffusion": { + "nodes": [ + { + "node_id": "n1", + "dependencies": [], + "method_label": "use nonnegativity of a square", + "instantiated_claim": "(x-2)^2 >= 0", + "justification": "squares are nonnegative", + }, + { + "node_id": "n2", + "dependencies": ["n1"], + "method_label": "expand the square", + "instantiated_claim": "x^2-4x+4 >= 0", + "justification": "expand n1", + }, + { + "node_id": "n3", + "dependencies": ["n2"], + "method_label": "divide by a positive quantity", + "instantiated_claim": "x+4/x >= 4", + "justification": "divide n2 by x>0", + }, + ], + "terminal_node_id": "n3", + "terminal_answer": "4", }, - f"{item_id}.candidate": { + f"{item_id}.stage5.render": { "question": "Let x>0. Prove that x+4/x >= 4.", "solution": ( - "Since (x-2)^2 >= 0, expansion and division by x>0 " - "give x+4/x >= 4." + "[n1] Since (x-2)^2 >= 0. [n2] Expanding gives " + "x^2-4x+4 >= 0. [n3] Divide by x>0 to get x+4/x >= 4." ), + "node_order": ["n1", "n2", "n3"], + "terminal_answer": "4", }, } ) @@ -191,11 +259,11 @@ async def run_offline_smoke(work_root: Path) -> dict[str, Any]: ) for _ in range(5) ] - kernel = await KernelPipeline( + kernel = await PaperKernelPipeline( proposer=proposer, judges=judges, store=RunStore(kernel_root, item_id), - config=PipelineConfig( + config=PaperPipelineConfig( proposer_model="scripted", judge_model="scripted", ), diff --git a/src/gap_pipeline/kernel_models.py b/src/gap_pipeline/kernel_models.py new file mode 100644 index 0000000..32abf39 --- /dev/null +++ b/src/gap_pipeline/kernel_models.py @@ -0,0 +1,348 @@ +"""Typed contracts for the literal five-stage GAP kernel pipeline.""" + +from __future__ import annotations + +import json +import re +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +from .models import SCHEMA_VERSION + + +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class ProofNode(StrictModel): + node_id: str + claim: str + dependencies: list[str] = Field(default_factory=list) + + @model_validator(mode="after") + def validate_content(self) -> "ProofNode": + if not re.fullmatch(r"n[1-9][0-9]*", self.node_id): + raise ValueError("proof node IDs must be n1, n2, ...") + if not self.claim.strip(): + raise ValueError("proof-node claim must be non-empty") + if len(self.dependencies) != len(set(self.dependencies)): + raise ValueError(f"{self.node_id} has duplicate dependencies") + return self + + +class ProofDAG(StrictModel): + nodes: list[ProofNode] = Field(min_length=1) + terminal_node_id: str + + def node_ids(self) -> list[str]: + return [node.node_id for node in self.nodes] + + @model_validator(mode="after") + def validate_graph(self) -> "ProofDAG": + node_ids = self.node_ids() + expected = [f"n{index}" for index in range(1, len(node_ids) + 1)] + if node_ids != expected: + raise ValueError("proof DAG nodes must be topologically ordered n1 through nN") + known: set[str] = set() + for node in self.nodes: + if node.node_id in node.dependencies: + raise ValueError(f"{node.node_id} cannot depend on itself") + unknown = set(node.dependencies) - known + if unknown: + raise ValueError( + f"{node.node_id} has non-prior dependencies {sorted(unknown)}" + ) + known.add(node.node_id) + if self.terminal_node_id not in known: + raise ValueError("terminal node is not present in the DAG") + + by_id = {node.node_id: node for node in self.nodes} + ancestors: set[str] = set() + pending = [self.terminal_node_id] + while pending: + node_id = pending.pop() + if node_id in ancestors: + continue + ancestors.add(node_id) + pending.extend(by_id[node_id].dependencies) + if ancestors != known: + raise ValueError("every proof node must contribute to the terminal node") + return self + + +class MethodNode(StrictModel): + node_id: str + method_label: str + + @model_validator(mode="after") + def validate_content(self) -> "MethodNode": + if not self.method_label.strip(): + raise ValueError("method label must be non-empty") + return self + + +class MethodPlan(StrictModel): + nodes: list[MethodNode] = Field(min_length=1) + + def validate_against(self, dag: ProofDAG) -> "MethodPlan": + if [node.node_id for node in self.nodes] != dag.node_ids(): + raise ValueError("method-plan IDs must exactly match proof-DAG IDs") + return self + + +class ReplacementChange(StrictModel): + slot_id: str + source_node_id: str + description: str + original_value: str + replacement_value: str + guard_condition: str + guard_justification: str + + @model_validator(mode="after") + def validate_change(self) -> "ReplacementChange": + text_fields = [ + self.slot_id, + self.source_node_id, + self.description, + self.original_value, + self.replacement_value, + self.guard_condition, + self.guard_justification, + ] + if any(not value.strip() for value in text_fields): + raise ValueError("replacement fields must be non-empty") + if self.original_value.strip() == self.replacement_value.strip(): + raise ValueError("replacement must differ from the original value") + return self + + +class ReplacementPlan(StrictModel): + changes: list[ReplacementChange] = Field(min_length=1) + closure_statement: str + + @model_validator(mode="after") + def validate_content(self) -> "ReplacementPlan": + slot_ids = [change.slot_id for change in self.changes] + if len(slot_ids) != len(set(slot_ids)): + raise ValueError("replacement slot IDs must be unique") + if not self.closure_statement.strip(): + raise ValueError("replacement plan must state its closure guarantee") + return self + + def validate_against(self, dag: ProofDAG) -> "ReplacementPlan": + known = set(dag.node_ids()) + unknown = { + change.source_node_id + for change in self.changes + if change.source_node_id not in known + } + if unknown: + raise ValueError(f"replacement plan references unknown nodes {sorted(unknown)}") + return self + + +class DiffusedProofNode(StrictModel): + node_id: str + dependencies: list[str] = Field(default_factory=list) + method_label: str + instantiated_claim: str + justification: str + + @model_validator(mode="after") + def validate_content(self) -> "DiffusedProofNode": + if not self.method_label.strip(): + raise ValueError("diffused method label must be non-empty") + if not self.instantiated_claim.strip() or not self.justification.strip(): + raise ValueError("diffused claim and justification must be non-empty") + return self + + +class DiffusedProof(StrictModel): + nodes: list[DiffusedProofNode] = Field(min_length=1) + terminal_node_id: str + terminal_answer: str + + def validate_against( + self, + dag: ProofDAG, + method_plan: MethodPlan, + ) -> "DiffusedProof": + method_plan.validate_against(dag) + if [node.node_id for node in self.nodes] != dag.node_ids(): + raise ValueError("diffused proof must contain exactly one row per DAG node") + dag_by_id = {node.node_id: node for node in dag.nodes} + methods = {node.node_id: node.method_label for node in method_plan.nodes} + for node in self.nodes: + if node.dependencies != dag_by_id[node.node_id].dependencies: + raise ValueError( + f"{node.node_id} dependencies changed during DAG diffusion" + ) + if node.method_label != methods[node.node_id]: + raise ValueError( + f"{node.node_id} method label changed during DAG diffusion" + ) + if self.terminal_node_id != dag.terminal_node_id: + raise ValueError("diffused terminal node must match the source DAG") + if not self.terminal_answer.strip(): + raise ValueError("diffused proof must expose a terminal answer") + return self + + +class RenderedVariant(StrictModel): + question: str + solution: str + node_order: list[str] = Field(min_length=1) + terminal_answer: str + + @model_validator(mode="after") + def validate_content(self) -> "RenderedVariant": + if not self.question.strip() or not self.solution.strip(): + raise ValueError("rendered question and solution must be non-empty") + if not self.terminal_answer.strip(): + raise ValueError("rendered terminal answer must be non-empty") + return self + + def validate_against( + self, + dag: ProofDAG, + diffused_proof: DiffusedProof, + ) -> "RenderedVariant": + if self.node_order != dag.node_ids(): + raise ValueError("rendered solution must cite every proof node in order") + missing_markers = [ + node_id + for node_id in self.node_order + if f"[{node_id}]" not in self.solution + ] + if missing_markers: + raise ValueError( + f"rendered solution is missing node markers {missing_markers}" + ) + if self.terminal_answer.strip() != diffused_proof.terminal_answer.strip(): + raise ValueError("rendered terminal answer differs from diffused proof") + return self + + +class CandidateBundle(StrictModel): + replacement_plan: ReplacementPlan + diffused_proof: DiffusedProof + variant: RenderedVariant + + +class JudgeVerdict(StrictModel): + verdict: Literal["accept", "reject"] + step_by_step_check: str + replacement_check: str + blocking_issues: str = "" + patch_suggestion: str = "" + + @field_validator( + "step_by_step_check", + "replacement_check", + "blocking_issues", + "patch_suggestion", + mode="before", + ) + @classmethod + def normalize_text_fields(cls, value: Any) -> str: + if value is None: + return "" + if isinstance(value, str): + return value + return json.dumps(value, ensure_ascii=False, sort_keys=True) + + @field_validator("blocking_issues", "patch_suggestion", mode="after") + @classmethod + def normalize_absence_sentinels(cls, value: str) -> str: + normalized = value.strip().lower().rstrip(".") + if normalized in { + "", + "none", + "none detected", + "n/a", + "no issues", + "no blocking issues", + }: + return "" + return value + + @model_validator(mode="after") + def validate_verdict(self) -> "JudgeVerdict": + if self.verdict == "accept" and self.blocking_issues.strip(): + raise ValueError("accept verdict cannot contain blocking issues") + if self.verdict == "reject" and not self.blocking_issues.strip(): + raise ValueError("reject verdict must identify a blocking issue") + return self + + def validate_coverage( + self, + dag: ProofDAG, + replacement_plan: ReplacementPlan, + ) -> "JudgeVerdict": + missing_nodes = [ + node_id + for node_id in dag.node_ids() + if re.search( + rf"(? "KernelRunResult": + if self.status == "accepted": + required = [ + self.accepted_replacement_plan, + self.accepted_diffused_proof, + self.accepted_candidate, + self.accepted_candidate_sha256, + self.accepted_bundle_sha256, + ] + if any(value is None for value in required): + raise ValueError("accepted run must include complete stage provenance") + elif not self.rejection_reason: + raise ValueError("rejected run must state a reason") + return self diff --git a/src/gap_pipeline/kernel_prompts.py b/src/gap_pipeline/kernel_prompts.py new file mode 100644 index 0000000..c4a6128 --- /dev/null +++ b/src/gap_pipeline/kernel_prompts.py @@ -0,0 +1,276 @@ +"""Prompts for an explicit implementation of the paper's five kernel stages. + +The byte-pinned historical Prompt-A/Prompt-B remain in ``prompts.py``. These +prompts make the richer five-stage manuscript description executable and keep +every transformation decision in a typed artifact. +""" + +from __future__ import annotations + +import json + +from .kernel_models import ( + DiffusedProof, + MethodPlan, + ProofDAG, + ReplacementPlan, +) +from .models import CanonicalItem + + +DAG_SYSTEM = "You are a rigorous competition-math proof analyst." +DAG_USER = """Parse the official solution into a concrete proof DAG. + +Each node must be an intermediate mathematical claim, not a method name. +Each dependency must be a local entailment used to establish that node. +List nodes in topological order with IDs n1, n2, ... . Every node must +contribute to one terminal node. + +Return JSON only: +{{"nodes":[{{"node_id":"n1","claim":"...","dependencies":[]}}], + "terminal_node_id":"nN"}} + +ORIGINAL PROBLEM: +<<<{question}>>> + +OFFICIAL SOLUTION: +<<<{solution}>>>""" + + +METHOD_SYSTEM = "You abstract concrete proofs into content-free method plans." +METHOD_USER = """For every concrete proof-DAG node, provide one content-free +method label. Preserve node IDs and order exactly. A method label describes the +operation, not the source constants, variable names, or final numerical answer. + +Return JSON only: +{{"nodes":[{{"node_id":"n1","method_label":"..."}}]}} + +PROOF DAG: +{dag}""" + + +REPLACEMENT_SYSTEM = "You design guarded, proof-plan-preserving math replacements." +REPLACEMENT_USER = """Choose one or more substantive numerical or structural +replacements that create a genuinely new problem while preserving the supplied +method plan. + +For every change: +- state the exact source DAG node; +- record the original and replacement values explicitly; +- state a mathematical guard condition derived from the original problem; +- explain why the replacement satisfies the guard. + +The list must be closed: stages 4 and 5 may introduce no mathematical change +that is not declared here. Variable renaming alone is not a kernel change. +{feedback} + +Return JSON only: +{{"changes":[ + {{"slot_id":"s1","source_node_id":"n1","description":"...", + "original_value":"...","replacement_value":"...", + "guard_condition":"...","guard_justification":"..."}} + ], + "closure_statement":"All intended mathematical changes are listed above."}} + +ORIGINAL PROBLEM: +<<<{question}>>> + +OFFICIAL SOLUTION: +<<<{solution}>>> + +CONCRETE PROOF DAG: +{dag} + +CONTENT-FREE METHOD PLAN: +{methods}""" + + +DIFFUSION_SYSTEM = "You re-instantiate a proof DAG under declared guarded replacements." +DIFFUSION_USER = """Propagate only the declared replacements through the proof +DAG, node by node. Return exactly one row for every source node, preserving its +ID, dependencies, and method label. Do not introduce undeclared constants, +objects, assumptions, reductions, or proof methods. + +Return JSON only: +{{"nodes":[ + {{"node_id":"n1","dependencies":[],"method_label":"...", + "instantiated_claim":"...","justification":"..."}} + ], + "terminal_node_id":"nN", + "terminal_answer":"..."}} + +ORIGINAL PROBLEM: +<<<{question}>>> + +SOURCE PROOF DAG: +{dag} + +METHOD PLAN: +{methods} + +DECLARED REPLACEMENTS: +{replacements}""" + + +RENDER_SYSTEM = "You render a verified proof DAG into a self-contained Putnam problem." +RENDER_USER = """Render the diffused proof into one complete problem statement +and one complete solution. + +Requirements: +- the question must be fully determined by the diffused terminal claim; +- the solution must follow the diffused nodes in order and visibly mark each + paragraph with [n1], [n2], ...; +- node_order must equal exactly {node_order}; +- do not add any mathematical change beyond the declared replacements; +- copy the terminal answer exactly. + +Return JSON only: +{{"question":"...","solution":"...","node_order":{node_order}, + "terminal_answer":"..."}} + +DECLARED REPLACEMENTS: +{replacements} + +DIFFUSED PROOF: +{diffused}""" + + +JUDGE_SYSTEM = """You are a verification judge for a literal five-stage GAP +kernel transformation. Verification, not de novo solving, is your task.""" +JUDGE_USER = """Check the candidate against every disclosed artifact. + +You must verify: +1. every replacement satisfies its guard and all mathematical changes are + declared in the replacement plan; +2. each diffused node instantiates the matching source node and method label; +3. dependencies and proof order are preserved; +4. the rendered problem is well-posed and the rendered solution proves it; +5. the terminal answer agrees across diffusion, rendering, and solution. + +In step_by_step_check, mention each of these node IDs explicitly: +{required_node_ids} +In replacement_check, mention each of these slot IDs explicitly: +{required_slot_ids} +{format_feedback} + +Return JSON only: +{{"verdict":"accept" or "reject", + "step_by_step_check":"...", + "replacement_check":"...", + "blocking_issues":"...", + "patch_suggestion":"..."}} + +ORIGINAL PROBLEM: +<<<{question}>>> + +OFFICIAL SOLUTION: +<<<{solution}>>> + +SOURCE PROOF DAG: +{dag} + +METHOD PLAN: +{methods} + +REPLACEMENT PLAN: +{replacements} + +DIFFUSED PROOF: +{diffused} + +RENDERED VARIANT: +{variant}""" + + +def _dump(value: object) -> str: + if hasattr(value, "model_dump"): + value = value.model_dump(mode="json") + return json.dumps(value, ensure_ascii=False, indent=2) + + +def dag_user(item: CanonicalItem) -> str: + return DAG_USER.format(question=item.problem, solution=item.solution) + + +def method_user(dag: ProofDAG) -> str: + return METHOD_USER.format(dag=_dump(dag)) + + +def replacement_user( + item: CanonicalItem, + dag: ProofDAG, + methods: MethodPlan, + *, + feedback: str = "", +) -> str: + feedback_block = ( + f"Previous verification feedback to address:\n{feedback}" + if feedback + else "This is the initial replacement proposal." + ) + return REPLACEMENT_USER.format( + feedback=feedback_block, + question=item.problem, + solution=item.solution, + dag=_dump(dag), + methods=_dump(methods), + ) + + +def diffusion_user( + item: CanonicalItem, + dag: ProofDAG, + methods: MethodPlan, + replacements: ReplacementPlan, +) -> str: + return DIFFUSION_USER.format( + question=item.problem, + dag=_dump(dag), + methods=_dump(methods), + replacements=_dump(replacements), + ) + + +def render_user( + replacements: ReplacementPlan, + diffused: DiffusedProof, +) -> str: + return RENDER_USER.format( + node_order=json.dumps( + [node.node_id for node in diffused.nodes], + ensure_ascii=False, + ), + replacements=_dump(replacements), + diffused=_dump(diffused), + ) + + +def judge_user( + item: CanonicalItem, + dag: ProofDAG, + methods: MethodPlan, + replacements: ReplacementPlan, + diffused: DiffusedProof, + variant: object, + *, + format_feedback: str = "", +) -> str: + return JUDGE_USER.format( + required_node_ids=json.dumps(dag.node_ids(), ensure_ascii=False), + required_slot_ids=json.dumps( + [change.slot_id for change in replacements.changes], + ensure_ascii=False, + ), + format_feedback=( + f"Previous report-format error: {format_feedback}" + if format_feedback + else "" + ), + question=item.problem, + solution=item.solution, + dag=_dump(dag), + methods=_dump(methods), + replacements=_dump(replacements), + diffused=_dump(diffused), + variant=_dump(variant), + ) diff --git a/src/gap_pipeline/models.py b/src/gap_pipeline/models.py index 6fab32f..9b0f214 100644 --- a/src/gap_pipeline/models.py +++ b/src/gap_pipeline/models.py @@ -2,11 +2,12 @@ from __future__ import annotations +import json import re from datetime import datetime, timezone from typing import Any, Literal -from pydantic import BaseModel, ConfigDict, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator SCHEMA_VERSION = "gap-prompt-faithful-v1" @@ -194,6 +195,20 @@ class JudgeVerdict(StrictModel): blocking_issues: str = "" patch_suggestion: str = "" + @field_validator( + "step_by_step_check", + "blocking_issues", + "patch_suggestion", + mode="before", + ) + @classmethod + def normalize_text_fields(cls, value: Any) -> str: + if value is None: + return "" + if isinstance(value, str): + return value + return json.dumps(value, ensure_ascii=False, sort_keys=True) + @model_validator(mode="after") def validate_verdict(self) -> "JudgeVerdict": if self.verdict == "accept" and self.blocking_issues.strip(): diff --git a/src/gap_pipeline/offline.py b/src/gap_pipeline/offline.py index 2be8922..e1c4927 100644 --- a/src/gap_pipeline/offline.py +++ b/src/gap_pipeline/offline.py @@ -198,7 +198,7 @@ def summarize_run(run_dir: Path) -> dict[str, Any]: first_round_failed += 1 if status == "accepted": accepted_ids.append(item_id) - if any(row.get("repaired", False) for row in history): + if any(not row.get("unanimous", False) for row in history): repaired_accepts += 1 else: rejected_ids.append(item_id) @@ -234,12 +234,12 @@ def align_run_to_release(run_dir: Path, dataset_dir: Path) -> dict[str, Any]: 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( + question_exact = candidate["question"] == released_kernel["question"] + solution_exact = candidate["solution"] == released_kernel["solution"] + question_normalized = normalize_text(candidate["question"]) == normalize_text( released_kernel["question"] ) - solution_normalized = normalize_text(candidate["proof"]) == normalize_text( + solution_normalized = normalize_text(candidate["solution"]) == normalize_text( released_kernel["solution"] ) rows.append( @@ -258,8 +258,8 @@ def align_run_to_release(run_dir: Path, dataset_dir: Path) -> dict[str, Any]: "solution_normalized": solution_normalized, "candidate_sha256": sha256_payload( { - "question": candidate["problem"], - "solution": candidate["proof"], + "question": candidate["question"], + "solution": candidate["solution"], } ), "release_sha256": sha256_payload( diff --git a/src/gap_pipeline/paper_pipeline.py b/src/gap_pipeline/paper_pipeline.py new file mode 100644 index 0000000..84f34bf --- /dev/null +++ b/src/gap_pipeline/paper_pipeline.py @@ -0,0 +1,395 @@ +"""Literal five-stage GAP kernel pipeline matching the manuscript operations.""" + +from __future__ import annotations + +import asyncio + +from pydantic import BaseModel, ConfigDict, model_validator + +from .clients import JsonLLM +from .kernel_models import ( + CandidateBundle, + DiffusedProof, + JudgeVerdict, + KernelRunResult, + MethodPlan, + ProofDAG, + RenderedVariant, + ReplacementPlan, + VerificationIteration, +) +from .kernel_prompts import ( + DAG_SYSTEM, + DIFFUSION_SYSTEM, + JUDGE_SYSTEM, + METHOD_SYSTEM, + RENDER_SYSTEM, + REPLACEMENT_SYSTEM, + dag_user, + diffusion_user, + judge_user, + method_user, + render_user, + replacement_user, +) +from .models import CanonicalItem, ModelCallRecord +from .store import RunStore, sha256_payload + + +class PaperPipelineConfig(BaseModel): + model_config = ConfigDict(extra="forbid") + + protocol_name: str = "gap-literal-five-stage-J5-K2-T15" + proposer_model: str + judge_model: str + judge_count: int = 5 + streak_length: int = 2 + max_iterations: int = 15 + + @model_validator(mode="after") + def enforce_protocol(self) -> "PaperPipelineConfig": + 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") + return self + + +class PaperKernelPipeline: + """Execute all five paper stages as typed, separately auditable calls.""" + + def __init__( + self, + *, + proposer: JsonLLM, + judges: list[JsonLLM], + store: RunStore, + config: PaperPipelineConfig, + ) -> None: + if len(judges) != config.judge_count: + raise ValueError( + f"expected {config.judge_count} judges, 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, + ) -> dict: + 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.data + + async def construct_dag(self, item: CanonicalItem) -> ProofDAG: + request_id = f"{item.item_id}.stage1.dag" + dag = ProofDAG.model_validate( + await self._call( + self.proposer, + request_id=request_id, + system_prompt=DAG_SYSTEM, + user_prompt=dag_user(item), + ) + ) + self.store.write_stage("01_proof_dag", dag, request_id=request_id) + return dag + + async def summarize_methods(self, dag: ProofDAG) -> MethodPlan: + request_id = f"{self.store.item_id}.stage2.methods" + methods = MethodPlan.model_validate( + await self._call( + self.proposer, + request_id=request_id, + system_prompt=METHOD_SYSTEM, + user_prompt=method_user(dag), + ) + ).validate_against(dag) + self.store.write_stage("02_method_plan", methods, request_id=request_id) + return methods + + async def generate_replacements( + self, + item: CanonicalItem, + dag: ProofDAG, + methods: MethodPlan, + *, + version: int, + feedback: str = "", + ) -> ReplacementPlan: + request_id = f"{item.item_id}.stage3.replacement.v{version:02d}" + replacements = ReplacementPlan.model_validate( + await self._call( + self.proposer, + request_id=request_id, + system_prompt=REPLACEMENT_SYSTEM, + user_prompt=replacement_user( + item, + dag, + methods, + feedback=feedback, + ), + ) + ).validate_against(dag) + self.store.write_stage( + f"03_replacement_v{version:02d}", + replacements, + request_id=request_id, + ) + return replacements + + async def diffuse_dag( + self, + item: CanonicalItem, + dag: ProofDAG, + methods: MethodPlan, + replacements: ReplacementPlan, + *, + version: int, + ) -> DiffusedProof: + request_id = f"{item.item_id}.stage4.diffusion.v{version:02d}" + diffused = DiffusedProof.model_validate( + await self._call( + self.proposer, + request_id=request_id, + system_prompt=DIFFUSION_SYSTEM, + user_prompt=diffusion_user(item, dag, methods, replacements), + ) + ).validate_against(dag, methods) + self.store.write_stage( + f"04_diffused_proof_v{version:02d}", + diffused, + request_id=request_id, + ) + return diffused + + async def render_variant( + self, + dag: ProofDAG, + replacements: ReplacementPlan, + diffused: DiffusedProof, + *, + version: int, + ) -> RenderedVariant: + request_id = f"{self.store.item_id}.stage5.render.v{version:02d}" + variant = RenderedVariant.model_validate( + await self._call( + self.proposer, + request_id=request_id, + system_prompt=RENDER_SYSTEM, + user_prompt=render_user(replacements, diffused), + ) + ).validate_against(dag, diffused) + self.store.write_stage( + f"05_rendered_variant_v{version:02d}", + variant, + request_id=request_id, + ) + return variant + + async def build_bundle( + self, + item: CanonicalItem, + dag: ProofDAG, + methods: MethodPlan, + *, + version: int, + feedback: str = "", + ) -> CandidateBundle: + replacements = await self.generate_replacements( + item, + dag, + methods, + version=version, + feedback=feedback, + ) + diffused = await self.diffuse_dag( + item, + dag, + methods, + replacements, + version=version, + ) + variant = await self.render_variant( + dag, + replacements, + diffused, + version=version, + ) + return CandidateBundle( + replacement_plan=replacements, + diffused_proof=diffused, + variant=variant, + ) + + async def _judge_once( + self, + judge: JsonLLM, + *, + item: CanonicalItem, + dag: ProofDAG, + methods: MethodPlan, + bundle: CandidateBundle, + iteration: int, + judge_id: int, + ) -> JudgeVerdict: + format_feedback = "" + for attempt in range(1, 4): + request_id = ( + f"{item.item_id}.verify.t{iteration:02d}." + f"j{judge_id}.a{attempt}" + ) + verdict = JudgeVerdict.model_validate( + await self._call( + judge, + request_id=request_id, + system_prompt=JUDGE_SYSTEM, + user_prompt=judge_user( + item, + dag, + methods, + bundle.replacement_plan, + bundle.diffused_proof, + bundle.variant, + format_feedback=format_feedback, + ), + ) + ) + try: + return verdict.validate_coverage(dag, bundle.replacement_plan) + except ValueError as exc: + format_feedback = str(exc) + raise ValueError( + f"judge {judge_id} failed coverage after three format attempts: " + f"{format_feedback}" + ) + + @staticmethod + def _feedback(verdicts: list[JudgeVerdict]) -> str: + rows = [] + for index, verdict in enumerate(verdicts, start=1): + if verdict.verdict == "reject": + rows.append( + f"judge {index}: {verdict.blocking_issues}; " + f"suggested repair: {verdict.patch_suggestion}" + ) + return "\n".join(rows) + + async def verify( + self, + item: CanonicalItem, + dag: ProofDAG, + methods: MethodPlan, + initial_bundle: CandidateBundle, + ) -> KernelRunResult: + bundle = initial_bundle + iterations: list[VerificationIteration] = [] + pass_streak = 0 + streak_sha: str | None = None + repaired_from_previous = False + + for iteration in range(1, self.config.max_iterations + 1): + bundle_sha = sha256_payload(bundle) + candidate_sha = sha256_payload(bundle.variant) + verdicts = list( + await asyncio.gather( + *( + self._judge_once( + judge, + item=item, + dag=dag, + methods=methods, + bundle=bundle, + 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_sha not in {None, bundle_sha}: + raise AssertionError("pass streak crossed provenance versions") + streak_sha = bundle_sha + pass_streak += 1 + else: + pass_streak = 0 + streak_sha = None + + record = VerificationIteration( + iteration=iteration, + bundle_sha256=bundle_sha, + candidate_sha256=candidate_sha, + verdicts=verdicts, + unanimous=unanimous, + pass_streak_after=pass_streak, + repaired_from_previous=repaired_from_previous, + ) + 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", + proof_dag=dag, + method_plan=methods, + accepted_replacement_plan=bundle.replacement_plan, + accepted_diffused_proof=bundle.diffused_proof, + accepted_candidate=bundle.variant, + iterations=iterations, + accepted_candidate_sha256=candidate_sha, + accepted_bundle_sha256=bundle_sha, + ) + self.store.write_final(result) + return result + + if not unanimous and iteration < self.config.max_iterations: + bundle = await self.build_bundle( + item, + dag, + methods, + version=iteration + 1, + feedback=self._feedback(verdicts), + ) + repaired_from_previous = True + else: + repaired_from_previous = False + + result = KernelRunResult( + item_id=item.item_id, + status="rejected", + proof_dag=dag, + method_plan=methods, + iterations=iterations, + rejection_reason="no two consecutive unanimous rounds within T=15", + ) + self.store.write_final(result) + return result + + async def run(self, item: CanonicalItem) -> KernelRunResult: + self.store.write_input(item) + self.store.write_config(self.config) + dag = await self.construct_dag(item) + methods = await self.summarize_methods(dag) + bundle = await self.build_bundle(item, dag, methods, version=1) + return await self.verify(item, dag, methods, bundle) diff --git a/src/gap_pipeline/pipeline.py b/src/gap_pipeline/pipeline.py index 76ae722..4a5332a 100644 --- a/src/gap_pipeline/pipeline.py +++ b/src/gap_pipeline/pipeline.py @@ -1,4 +1,7 @@ -"""Prompt-faithful kernel generation and the J=5, K=2, T=15 loop.""" +"""Historical two-call Putnam generator retained for provenance tests. + +The default manuscript-aligned implementation is ``paper_pipeline.py``. +""" from __future__ import annotations diff --git a/src/gap_pipeline/release.py b/src/gap_pipeline/release.py index 9afa117..7df6566 100644 --- a/src/gap_pipeline/release.py +++ b/src/gap_pipeline/release.py @@ -92,6 +92,19 @@ def export_release( variants["kernel_variant"] = { "question": candidate["question"], "solution": candidate["solution"], + "_meta": { + "proof_dag": kernel_payload["proof_dag"], + "method_plan": kernel_payload["method_plan"], + "replacement_plan": kernel_payload["accepted_replacement_plan"], + "diffused_proof": kernel_payload["accepted_diffused_proof"], + "terminal_answer": candidate["terminal_answer"], + "accepted_candidate_sha256": kernel_payload[ + "accepted_candidate_sha256" + ], + "accepted_bundle_sha256": kernel_payload[ + "accepted_bundle_sha256" + ], + }, } output_record = copy.deepcopy(record) output_record["variants"] = variants diff --git a/src/gap_pipeline/surface.py b/src/gap_pipeline/surface.py index 914d847..aae34a8 100644 --- a/src/gap_pipeline/surface.py +++ b/src/gap_pipeline/surface.py @@ -29,6 +29,24 @@ SURFACE_FAMILIES: tuple[SurfaceFamily, ...] = ( IDENTIFIER_RE = re.compile(r"^[a-z]{8,}$") +def apply_rename_map(text: str, rename_map: dict[str, str]) -> str: + """Apply symbol renames without replacing letters inside other tokens.""" + + rendered = text + for source in sorted(rename_map, key=len, reverse=True): + pattern = re.escape(source) + if source[0].isalnum(): + pattern = rf"(?