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
138
|
from __future__ import annotations
import pytest
from gap_pipeline.kernel_models import (
DiffusedProof,
JudgeVerdict,
MethodPlan,
ProofDAG,
RenderedVariant,
ReplacementPlan,
)
def branched_dag() -> ProofDAG:
return ProofDAG.model_validate(
{
"nodes": [
{"node_id": "n1", "claim": "first fact", "dependencies": []},
{"node_id": "n2", "claim": "left branch", "dependencies": ["n1"]},
{"node_id": "n3", "claim": "right branch", "dependencies": ["n1"]},
{
"node_id": "n4",
"claim": "combine branches",
"dependencies": ["n2", "n3"],
},
],
"terminal_node_id": "n4",
}
)
def test_proof_dag_supports_real_branching() -> None:
dag = branched_dag()
assert dag.nodes[-1].dependencies == ["n2", "n3"]
def test_replacement_must_reference_a_real_dag_node() -> None:
replacements = ReplacementPlan.model_validate(
{
"changes": [
{
"slot_id": "s1",
"source_node_id": "n9",
"description": "constant",
"original_value": "1",
"replacement_value": "2",
"guard_condition": "positive",
"guard_justification": "2 is positive",
}
],
"closure_statement": "No undeclared changes.",
}
)
with pytest.raises(ValueError, match="unknown nodes"):
replacements.validate_against(branched_dag())
def test_diffusion_preserves_dependencies_and_methods() -> None:
dag = branched_dag()
methods = MethodPlan.model_validate(
{
"nodes": [
{"node_id": node.node_id, "method_label": f"method {node.node_id}"}
for node in dag.nodes
]
}
)
diffused = DiffusedProof.model_validate(
{
"nodes": [
{
"node_id": node.node_id,
"dependencies": node.dependencies,
"method_label": f"method {node.node_id}",
"instantiated_claim": f"new {node.claim}",
"justification": "valid re-instantiation",
}
for node in dag.nodes
],
"terminal_node_id": "n4",
"terminal_answer": "answer",
}
)
assert diffused.validate_against(dag, methods) is diffused
broken = diffused.model_copy(deep=True)
broken.nodes[-1].dependencies = ["n3"]
with pytest.raises(ValueError, match="dependencies changed"):
broken.validate_against(dag, methods)
def test_rendered_solution_must_expose_every_node() -> None:
dag = branched_dag()
methods = MethodPlan.model_validate(
{
"nodes": [
{"node_id": node.node_id, "method_label": f"method {node.node_id}"}
for node in dag.nodes
]
}
)
diffused = DiffusedProof.model_validate(
{
"nodes": [
{
"node_id": node.node_id,
"dependencies": node.dependencies,
"method_label": f"method {node.node_id}",
"instantiated_claim": f"new {node.claim}",
"justification": "valid",
}
for node in dag.nodes
],
"terminal_node_id": "n4",
"terminal_answer": "answer",
}
).validate_against(dag, methods)
variant = RenderedVariant(
question="New problem",
solution="[n1] first [n2] left [n3] right",
node_order=dag.node_ids(),
terminal_answer="answer",
)
with pytest.raises(ValueError, match="missing node markers.*n4"):
variant.validate_against(dag, diffused)
def test_accept_verdict_normalizes_explicit_none_sentinel() -> None:
verdict = JudgeVerdict(
verdict="accept",
step_by_step_check="n1 valid",
replacement_check="s1 valid",
blocking_issues="None detected.",
patch_suggestion="N/A",
)
assert verdict.blocking_issues == ""
assert verdict.patch_suggestion == ""
|
