summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/gap_pipeline/cli.py6
-rw-r--r--src/gap_pipeline/e2e.py104
-rw-r--r--src/gap_pipeline/kernel_models.py348
-rw-r--r--src/gap_pipeline/kernel_prompts.py276
-rw-r--r--src/gap_pipeline/models.py17
-rw-r--r--src/gap_pipeline/offline.py14
-rw-r--r--src/gap_pipeline/paper_pipeline.py395
-rw-r--r--src/gap_pipeline/pipeline.py5
-rw-r--r--src/gap_pipeline/release.py13
-rw-r--r--src/gap_pipeline/surface.py29
10 files changed, 1175 insertions, 32 deletions
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"(?<![A-Za-z0-9_]){re.escape(node_id)}(?![A-Za-z0-9_])",
+ self.step_by_step_check,
+ )
+ is None
+ ]
+ missing_slots = [
+ change.slot_id
+ for change in replacement_plan.changes
+ if re.search(
+ rf"(?<![A-Za-z0-9_]){re.escape(change.slot_id)}(?![A-Za-z0-9_])",
+ self.replacement_check,
+ )
+ is None
+ ]
+ if missing_nodes or missing_slots:
+ raise ValueError(
+ "judge coverage incomplete: "
+ f"nodes={missing_nodes}, slots={missing_slots}"
+ )
+ return self
+
+
+class VerificationIteration(StrictModel):
+ iteration: int
+ bundle_sha256: str
+ candidate_sha256: str
+ verdicts: list[JudgeVerdict]
+ unanimous: bool
+ pass_streak_after: int
+ repaired_from_previous: bool = False
+
+
+class KernelRunResult(StrictModel):
+ item_id: str
+ status: Literal["accepted", "rejected"]
+ proof_dag: ProofDAG
+ method_plan: MethodPlan
+ accepted_replacement_plan: ReplacementPlan | None = None
+ accepted_diffused_proof: DiffusedProof | None = None
+ accepted_candidate: RenderedVariant | None = None
+ iterations: list[VerificationIteration]
+ rejection_reason: str = ""
+ accepted_candidate_sha256: str | None = None
+ accepted_bundle_sha256: str | None = None
+ schema_version: str = f"{SCHEMA_VERSION}-literal-five-stage"
+
+ @model_validator(mode="after")
+ def validate_terminal_state(self) -> "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"(?<![A-Za-z0-9_]){pattern}"
+ if source[-1].isalnum():
+ pattern = rf"{pattern}(?![A-Za-z0-9_])"
+ rendered = re.sub(
+ pattern,
+ lambda _match, replacement=rename_map[source]: replacement,
+ rendered,
+ )
+ return rendered
+
+
@dataclass(frozen=True)
class SurfaceVariant:
family: SurfaceFamily
@@ -69,6 +87,13 @@ def validate_surface_variant(
)
if not question.strip() or not solution.strip():
raise ValueError("surface question and solution must be non-empty")
+ expected_question = apply_rename_map(item.problem, rename_map)
+ expected_solution = apply_rename_map(item.solution, rename_map)
+ if question != expected_question or solution != expected_solution:
+ raise ValueError(
+ "surface text must be the exact deterministic rename of the "
+ "canonical question and solution"
+ )
class SurfacePipeline:
@@ -105,8 +130,8 @@ class SurfacePipeline:
str(old): str(new)
for old, new in dict(response.data["map"]).items()
}
- question = str(response.data["question"])
- solution = str(response.data["solution"])
+ question = apply_rename_map(item.problem, rename_map)
+ solution = apply_rename_map(item.solution, rename_map)
validate_surface_variant(
item,
rename_map=rename_map,
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-08 12:58:18 +0000