1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
from __future__ import annotations
import pytest
from pydantic import ValidationError
from gap_pipeline.models import (
JudgeVerdict,
KernelCandidate,
KernelPlan,
ProofPlanDAG,
)
def test_valid_plan_and_candidate(plan_dict: dict, candidate_dict: dict) -> None:
plan = KernelPlan.model_validate(plan_dict)
candidate = KernelCandidate.model_validate(candidate_dict)
assert len(plan.core_steps) == 2
assert candidate.question.startswith("Let x")
def test_plan_requires_slots(plan_dict: dict) -> None:
plan_dict["mutable_slots"] = {}
with pytest.raises(ValidationError, match="mutable slot"):
KernelPlan.model_validate(plan_dict)
def test_plan_rejects_more_than_five_core_steps(plan_dict: dict) -> None:
plan_dict["core_steps"] = [f"step {index}" for index in range(6)]
with pytest.raises(ValidationError):
KernelPlan.model_validate(plan_dict)
def test_candidate_requires_question_and_solution(candidate_dict: dict) -> None:
candidate_dict["solution"] = ""
with pytest.raises(ValidationError, match="non-empty"):
KernelCandidate.model_validate(candidate_dict)
def test_accept_verdict_cannot_report_blocking_issue() -> None:
with pytest.raises(ValidationError, match="cannot contain"):
JudgeVerdict(
verdict="accept",
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_does_not_require_keyword_coverage() -> None:
verdict = JudgeVerdict(
verdict="reject",
step_by_step_check=(
"The early derivation is correct, but the announced replacement "
"never appears in the candidate problem."
),
blocking_issues="the candidate is unchanged from the original",
patch_suggestion="apply the declared replacement",
)
assert verdict.verdict == "reject"
def test_judge_verdict_normalizes_structured_text_fields() -> None:
verdict = JudgeVerdict.model_validate(
{
"verdict": "accept",
"step_by_step_check": {
"n1": "correctly instantiated",
"n2": "correctly instantiated",
},
"blocking_issues": None,
"patch_suggestion": None,
}
)
assert '"n1"' in verdict.step_by_step_check
assert verdict.blocking_issues == ""
assert verdict.patch_suggestion == ""
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)
|
