Validate path-structured proof-plan DAGHEADmain

9 files changed, 255 insertions, 25 deletions

diff --git a/README.md b/README.md
index 090c297..b1a3f8c 100644
--- a/README.md
+++ b/README.md
@@ -43,6 +43,14 @@ The conceptual five stages are represented explicitly in saved artifacts.
 The original implementation batches stages 1–3 into Prompt-A and stages 4–5
 into Prompt-B; the wrapper does not change those prompts. See `STAGE_MAP.md`.
 
+Prompt-A's ordered core steps instantiate a path-structured proof-plan DAG:
+each step is a typed node and each edge records the dependency on the preceding
+step. `ProofPlanDAG` validates unique node IDs, known dependencies, acyclicity,
+connectivity to the terminal node, and the extracted order. This path structure
+is the precise graph induced by an ordered minimal proof chain. Judges receive
+the same method-label sequence with stable node IDs and must report a check for
+every node before their verdict is counted.
+
 ## Install and test
 
 ```bash
diff --git a/STAGE_MAP.md b/STAGE_MAP.md
index 26077a3..ee77f96 100644
--- a/STAGE_MAP.md
+++ b/STAGE_MAP.md
@@ -6,10 +6,16 @@ re-instantiates that plan and returns the regenerated proof and question.
 The package writes each conceptual output separately so every paper stage is
 visible in an end-to-end run without changing either prompt.
 
+The ordered nodes form a path-structured DAG, the exact graph induced by a
+minimal sequential proof plan. `ProofPlanDAG` validates node identity,
+dependencies, acyclicity, terminal connectivity, and order. The five judges
+receive stable node IDs with the method-label sequence, and a verdict is valid
+only when its step-by-step check covers every node.
+
 | Paper operation | Implementation | Saved artifact |
 |---|---|---|
 | Surface rename | `SurfacePipeline.run_family` | `surface_<family>_variant.json` |
-| 1. Reference solution to proof structure | `KernelPipeline.extract_plan` | `01_proof_dag.json` |
+| 1. Reference solution to proof structure | `KernelPipeline.extract_plan` + `ProofPlanDAG` validation | `01_proof_dag.json` |
 | 2. Content-free method plan | `KernelPipeline.extract_plan` | `02_method_plan.json` |
 | 3. Mutable-slot identification | `KernelPipeline.extract_plan` | `03_mutable_slots.json` |
 | 4. Proof regeneration | `KernelPipeline.generate_candidate` | `04_regenerated_proof.json` |
diff --git a/src/gap_pipeline/__init__.py b/src/gap_pipeline/__init__.py
index fe2e04c..fbc2adb 100644
--- a/src/gap_pipeline/__init__.py
+++ b/src/gap_pipeline/__init__.py
@@ -5,6 +5,8 @@ from .models import (
     KernelCandidate,
     KernelPlan,
     KernelRunResult,
+    ProofPlanDAG,
+    ProofPlanNode,
 )
 from .pipeline import KernelPipeline, PipelineConfig
 
@@ -15,6 +17,8 @@ __all__ = [
     "KernelPlan",
     "KernelRunResult",
     "PipelineConfig",
+    "ProofPlanDAG",
+    "ProofPlanNode",
 ]
 
 __version__ = "0.2.0"
diff --git a/src/gap_pipeline/e2e.py b/src/gap_pipeline/e2e.py
index 971da3e..ded1381 100644
--- a/src/gap_pipeline/e2e.py
+++ b/src/gap_pipeline/e2e.py
@@ -25,7 +25,7 @@ def _ensure_fresh(path: Path) -> None:
 def _review_accept() -> dict[str, str]:
     return {
         "verdict": "accept",
-        "step_by_step_check": "both method labels are instantiated",
+        "step_by_step_check": "n1 passes; n2 passes",
         "blocking_issues": "",
         "patch_suggestion": "",
     }
diff --git a/src/gap_pipeline/models.py b/src/gap_pipeline/models.py
index bf80c5b..6fab32f 100644
--- a/src/gap_pipeline/models.py
+++ b/src/gap_pipeline/models.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import re
 from datetime import datetime, timezone
 from typing import Any, Literal
 
@@ -67,6 +68,115 @@ class KernelPlan(StrictModel):
         return self
 
 
+class ProofPlanNode(StrictModel):
+    node_id: str
+    method_label: str
+    dependencies: list[str] = Field(default_factory=list)
+
+    @model_validator(mode="after")
+    def validate_content(self) -> "ProofPlanNode":
+        if not self.node_id.strip():
+            raise ValueError("proof-plan node ID must be non-empty")
+        if not self.method_label.strip():
+            raise ValueError("proof-plan method label must be non-empty")
+        return self
+
+
+class ProofPlanDAG(StrictModel):
+    """Typed path-DAG induced by Prompt-A's ordered proof-plan steps."""
+
+    nodes: list[ProofPlanNode] = Field(min_length=1)
+    terminal_node_id: str
+    structure: Literal["path"] = "path"
+
+    @classmethod
+    def from_plan(cls, plan: KernelPlan) -> "ProofPlanDAG":
+        nodes = [
+            ProofPlanNode(
+                node_id=f"n{index}",
+                method_label=step,
+                dependencies=[] if index == 1 else [f"n{index - 1}"],
+            )
+            for index, step in enumerate(plan.core_steps, start=1)
+        ]
+        return cls(nodes=nodes, terminal_node_id=nodes[-1].node_id)
+
+    def method_labels_payload(self) -> list[str]:
+        """Keep the judge input a sequence while making node coverage auditable."""
+
+        return [f"{node.node_id}: {node.method_label}" for node in self.nodes]
+
+    def validate_plan(self, plan: KernelPlan) -> "ProofPlanDAG":
+        labels = [node.method_label for node in self.nodes]
+        if labels != plan.core_steps:
+            raise ValueError(
+                "proof-plan DAG labels must equal Prompt-A core_steps in order"
+            )
+        return self
+
+    @model_validator(mode="after")
+    def validate_graph(self) -> "ProofPlanDAG":
+        node_ids = [node.node_id for node in self.nodes]
+        if len(node_ids) != len(set(node_ids)):
+            raise ValueError("proof-plan DAG node IDs must be unique")
+
+        known = set(node_ids)
+        if self.terminal_node_id not in known:
+            raise ValueError("proof-plan DAG terminal node is unknown")
+
+        dependents: dict[str, list[str]] = {node_id: [] for node_id in node_ids}
+        indegree = {node.node_id: len(node.dependencies) for node in self.nodes}
+        for node in self.nodes:
+            if len(node.dependencies) != len(set(node.dependencies)):
+                raise ValueError(f"{node.node_id} has duplicate dependencies")
+            for dependency in node.dependencies:
+                if dependency not in known:
+                    raise ValueError(
+                        f"{node.node_id} depends on unknown node {dependency}"
+                    )
+                if dependency == node.node_id:
+                    raise ValueError(f"{node.node_id} cannot depend on itself")
+                dependents[dependency].append(node.node_id)
+
+        ready = [node_id for node_id in node_ids if indegree[node_id] == 0]
+        visited = 0
+        while ready:
+            node_id = ready.pop()
+            visited += 1
+            for dependent in dependents[node_id]:
+                indegree[dependent] -= 1
+                if indegree[dependent] == 0:
+                    ready.append(dependent)
+        if visited != len(node_ids):
+            raise ValueError("proof-plan graph contains a cycle")
+
+        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-plan node must lead to the terminal node")
+
+        expected_ids = [f"n{index}" for index in range(1, len(self.nodes) + 1)]
+        if node_ids != expected_ids:
+            raise ValueError("path-DAG nodes must be ordered n1 through nN")
+        for index, node in enumerate(self.nodes):
+            expected_dependencies = [] if index == 0 else [node_ids[index - 1]]
+            if node.dependencies != expected_dependencies:
+                raise ValueError(
+                    f"{node.node_id} must depend exactly on "
+                    f"{expected_dependencies or 'no prior node'}"
+                )
+        if self.terminal_node_id != node_ids[-1]:
+            raise ValueError("path-DAG terminal must be the final ordered node")
+        return self
+
+
 class KernelCandidate(StrictModel):
     question: str
     solution: str
@@ -92,6 +202,20 @@ class JudgeVerdict(StrictModel):
             raise ValueError("reject verdict must identify a blocking issue")
         return self
 
+    def validate_coverage(self, dag: ProofPlanDAG) -> "JudgeVerdict":
+        missing = [
+            node.node_id
+            for node in dag.nodes
+            if re.search(
+                rf"(?<![A-Za-z0-9_]){re.escape(node.node_id)}(?![A-Za-z0-9_])",
+                self.step_by_step_check,
+            )
+            is None
+        ]
+        if missing:
+            raise ValueError(f"judge check does not cover DAG nodes: {missing}")
+        return self
+
 
 class IterationRecord(StrictModel):
     iteration: int
@@ -105,6 +229,7 @@ class KernelRunResult(StrictModel):
     item_id: str
     status: Literal["accepted", "rejected"]
     plan: KernelPlan
+    proof_dag: ProofPlanDAG
     accepted_candidate: KernelCandidate | None = None
     iterations: list[IterationRecord]
     rejection_reason: str = ""
@@ -113,6 +238,7 @@ class KernelRunResult(StrictModel):
 
     @model_validator(mode="after")
     def validate_terminal_state(self) -> "KernelRunResult":
+        self.proof_dag.validate_plan(self.plan)
         if self.status == "accepted" and self.accepted_candidate is None:
             raise ValueError("accepted run must include its candidate")
         if self.status == "rejected" and not self.rejection_reason:
diff --git a/src/gap_pipeline/pipeline.py b/src/gap_pipeline/pipeline.py
index 307ba81..76ae722 100644
--- a/src/gap_pipeline/pipeline.py
+++ b/src/gap_pipeline/pipeline.py
@@ -15,6 +15,7 @@ from .models import (
     KernelPlan,
     KernelRunResult,
     ModelCallRecord,
+    ProofPlanDAG,
 )
 from .prompts import (
     FIX_SYSTEM_PROMPT,
@@ -91,7 +92,10 @@ class KernelPipeline:
         )
         return response.data
 
-    async def extract_plan(self, item: CanonicalItem) -> KernelPlan:
+    async def extract_plan(
+        self,
+        item: CanonicalItem,
+    ) -> tuple[KernelPlan, ProofPlanDAG]:
         request_id = f"{item.item_id}.plan"
         payload = await self._call(
             self.proposer,
@@ -100,21 +104,10 @@ class KernelPipeline:
             user_prompt=kernel_plan_user(item),
         )
         plan = KernelPlan.model_validate(payload)
-        nodes = [
-            {
-                "node_id": f"n{index}",
-                "method_label": step,
-                "dependencies": [] if index == 1 else [f"n{index - 1}"],
-            }
-            for index, step in enumerate(plan.core_steps, start=1)
-        ]
+        dag = ProofPlanDAG.from_plan(plan)
         self.store.write_stage(
             "01_proof_dag",
-            {
-                "nodes": nodes,
-                "terminal_node_id": nodes[-1]["node_id"],
-                "construction": "ordered core_steps returned verbatim by Prompt-A",
-            },
+            dag,
             request_id=request_id,
         )
         self.store.write_stage(
@@ -130,7 +123,7 @@ class KernelPipeline:
             },
             request_id=request_id,
         )
-        return plan
+        return plan, dag
 
     async def generate_candidate(
         self,
@@ -163,6 +156,7 @@ class KernelPipeline:
         *,
         item: CanonicalItem,
         plan: KernelPlan,
+        dag: ProofPlanDAG,
         candidate: KernelCandidate,
         iteration: int,
         judge_id: int,
@@ -172,9 +166,9 @@ class KernelPipeline:
             judge,
             request_id=request_id,
             system_prompt=JUDGE_SYSTEM_PROMPT,
-            user_prompt=judge_user(item, plan, candidate),
+            user_prompt=judge_user(item, plan, dag, candidate),
         )
-        return JudgeVerdict.model_validate(payload)
+        return JudgeVerdict.model_validate(payload).validate_coverage(dag)
 
     async def _repair(
         self,
@@ -220,8 +214,10 @@ class KernelPipeline:
         *,
         item: CanonicalItem,
         plan: KernelPlan,
+        dag: ProofPlanDAG,
         candidate: KernelCandidate,
     ) -> KernelRunResult:
+        dag.validate_plan(plan)
         current = candidate
         iterations: list[IterationRecord] = []
         pass_streak = 0
@@ -236,6 +232,7 @@ class KernelPipeline:
                             judge,
                             item=item,
                             plan=plan,
+                            dag=dag,
                             candidate=current,
                             iteration=iteration,
                             judge_id=judge_id,
@@ -269,6 +266,7 @@ class KernelPipeline:
                     item_id=item.item_id,
                     status="accepted",
                     plan=plan,
+                    proof_dag=dag,
                     accepted_candidate=current,
                     iterations=iterations,
                     accepted_candidate_sha256=current_sha,
@@ -288,6 +286,7 @@ class KernelPipeline:
             item_id=item.item_id,
             status="rejected",
             plan=plan,
+            proof_dag=dag,
             iterations=iterations,
             rejection_reason="no two consecutive unanimous rounds within T=15",
         )
@@ -297,10 +296,11 @@ class KernelPipeline:
     async def run(self, item: CanonicalItem) -> KernelRunResult:
         self.store.write_input(item)
         self.store.write_config(self.config)
-        plan = await self.extract_plan(item)
+        plan, dag = await self.extract_plan(item)
         candidate = await self.generate_candidate(item, plan)
         return await self.verify_candidate(
             item=item,
             plan=plan,
+            dag=dag,
             candidate=candidate,
         )
diff --git a/src/gap_pipeline/prompts.py b/src/gap_pipeline/prompts.py
index 2ba87cf..30d65b2 100644
--- a/src/gap_pipeline/prompts.py
+++ b/src/gap_pipeline/prompts.py
@@ -8,7 +8,7 @@ from __future__ import annotations
 
 import json
 
-from .models import CanonicalItem, KernelCandidate, KernelPlan
+from .models import CanonicalItem, KernelCandidate, KernelPlan, ProofPlanDAG
 
 
 # Source: PutnamVariants@c3bed737370df2dbf73afd66bf6e86d4ece82d68
@@ -220,12 +220,13 @@ def kernel_generate_user(item: CanonicalItem, plan: KernelPlan) -> str:
 def judge_user(
     item: CanonicalItem,
     plan: KernelPlan,
+    dag: ProofPlanDAG,
     candidate: KernelCandidate,
 ) -> str:
     return JUDGE_USER_TEMPLATE.format(
         original_problem=item.problem,
         original_solution=item.solution,
-        method_labels=json.dumps(plan.core_steps, ensure_ascii=False),
+        method_labels=json.dumps(dag.method_labels_payload(), ensure_ascii=False),
         slot_replacement=json.dumps(
             {
                 key: value.model_dump(mode="json")
diff --git a/tests/test_models.py b/tests/test_models.py
index 71dc854..d19b536 100644
--- a/tests/test_models.py
+++ b/tests/test_models.py
@@ -3,7 +3,12 @@ from __future__ import annotations
 import pytest
 from pydantic import ValidationError
 
-from gap_pipeline.models import JudgeVerdict, KernelCandidate, KernelPlan
+from gap_pipeline.models import (
+    JudgeVerdict,
+    KernelCandidate,
+    KernelPlan,
+    ProofPlanDAG,
+)
 
 
 def test_valid_plan_and_candidate(plan_dict: dict, candidate_dict: dict) -> None:
@@ -38,3 +43,79 @@ def test_accept_verdict_cannot_report_blocking_issue() -> None:
             step_by_step_check="checked",
             blocking_issues="an error",
         )
+
+
+def test_path_dag_is_derived_from_ordered_plan(plan_dict: dict) -> None:
+    dag = ProofPlanDAG.from_plan(KernelPlan.model_validate(plan_dict))
+    assert [node.node_id for node in dag.nodes] == ["n1", "n2"]
+    assert dag.nodes[0].dependencies == []
+    assert dag.nodes[1].dependencies == ["n1"]
+    assert dag.terminal_node_id == "n2"
+    assert dag.method_labels_payload()[0].startswith("n1: ")
+
+
+def test_dag_rejects_cycle() -> None:
+    with pytest.raises(ValidationError, match="cycle"):
+        ProofPlanDAG.model_validate(
+            {
+                "nodes": [
+                    {
+                        "node_id": "n1",
+                        "method_label": "first",
+                        "dependencies": ["n2"],
+                    },
+                    {
+                        "node_id": "n2",
+                        "method_label": "second",
+                        "dependencies": ["n1"],
+                    },
+                ],
+                "terminal_node_id": "n2",
+            }
+        )
+
+
+def test_path_dag_rejects_skipped_dependency() -> None:
+    with pytest.raises(ValidationError, match="lead to the terminal"):
+        ProofPlanDAG.model_validate(
+            {
+                "nodes": [
+                    {
+                        "node_id": "n1",
+                        "method_label": "first",
+                        "dependencies": [],
+                    },
+                    {
+                        "node_id": "n2",
+                        "method_label": "second",
+                        "dependencies": [],
+                    },
+                ],
+                "terminal_node_id": "n2",
+            }
+        )
+
+
+def test_judge_verdict_requires_every_dag_node(plan_dict: dict) -> None:
+    dag = ProofPlanDAG.from_plan(KernelPlan.model_validate(plan_dict))
+    verdict = JudgeVerdict(
+        verdict="accept",
+        step_by_step_check="n1 is valid; n2 is valid",
+    )
+    assert verdict.validate_coverage(dag) is verdict
+
+    with pytest.raises(ValueError, match="does not cover DAG nodes.*n2"):
+        JudgeVerdict(
+            verdict="accept",
+            step_by_step_check="n1 is valid",
+        ).validate_coverage(dag)
+
+
+def test_dag_labels_must_match_prompt_a_plan(plan_dict: dict) -> None:
+    plan = KernelPlan.model_validate(plan_dict)
+    dag = ProofPlanDAG.from_plan(plan)
+    changed_plan = plan.model_copy(
+        update={"core_steps": ["a different method", plan.core_steps[1]]}
+    )
+    with pytest.raises(ValueError, match="must equal Prompt-A core_steps"):
+        dag.validate_plan(changed_plan)
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index 864e967..04c3ea9 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -13,7 +13,7 @@ from conftest import changed_candidate
 def accept_review() -> dict[str, str]:
     return {
         "verdict": "accept",
-        "step_by_step_check": "every method label is instantiated",
+        "step_by_step_check": "n1 passes; n2 passes",
         "blocking_issues": "",
         "patch_suggestion": "",
     }
@@ -22,7 +22,9 @@ def accept_review() -> dict[str, str]:
 def reject_review(iteration: int) -> dict[str, str]:
     return {
         "verdict": "reject",
-        "step_by_step_check": f"terminal step fails at iteration {iteration}",
+        "step_by_step_check": (
+            f"n1 passes; n2 terminal step fails at iteration {iteration}"
+        ),
         "blocking_issues": "terminal algebra is incorrect",
         "patch_suggestion": "correct the terminal algebra",
     }
@@ -76,6 +78,8 @@ def test_two_consecutive_unanimous_passes_accept(
     final_path = tmp_path / "items" / item.item_id / "final.json"
     final = json.loads(final_path.read_text())
     assert final["accepted_candidate_sha256"]
+    assert final["proof_dag"]["structure"] == "path"
+    assert final["proof_dag"]["terminal_node_id"] == "n2"
     assert len(list((final_path.parent / "calls").glob("*.json"))) == 12