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
139
140
141
142
143
144
145
146
|
"""Validated schemas for the prompt-faithful GAP reproduction path."""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
SCHEMA_VERSION = "gap-prompt-faithful-v1"
class StrictModel(BaseModel):
model_config = ConfigDict(extra="forbid")
class CanonicalItem(StrictModel):
item_id: str
problem: str
solution: str
problem_type: Literal["proof", "calculation"] | str = "proof"
variables: list[str] = Field(default_factory=list)
parameters: list[str] = Field(default_factory=list)
scientific_constants: list[str] = Field(default_factory=list)
metadata: dict[str, Any] = Field(default_factory=dict)
@classmethod
def from_public_record(cls, record: dict[str, Any]) -> "CanonicalItem":
excluded = {
"index",
"question",
"solution",
"problem_type",
"vars",
"params",
"sci_consts",
"variants",
}
return cls(
item_id=str(record["index"]),
problem=str(record["question"]),
solution=str(record["solution"]),
problem_type=str(record.get("problem_type", "proof")),
variables=list(record.get("vars", [])),
parameters=list(record.get("params", [])),
scientific_constants=list(record.get("sci_consts", [])),
metadata={key: value for key, value in record.items() if key not in excluded},
)
class MutableSlot(StrictModel):
description: str
original: str
class KernelPlan(StrictModel):
core_steps: list[str] = Field(min_length=1, max_length=5)
mutable_slots: dict[str, MutableSlot]
@model_validator(mode="after")
def validate_content(self) -> "KernelPlan":
if any(not step.strip() for step in self.core_steps):
raise ValueError("core steps must be non-empty")
if not self.mutable_slots:
raise ValueError("at least one mutable slot is required")
return self
class KernelCandidate(StrictModel):
question: str
solution: str
@model_validator(mode="after")
def validate_content(self) -> "KernelCandidate":
if not self.question.strip() or not self.solution.strip():
raise ValueError("kernel question and solution must be non-empty")
return self
class JudgeVerdict(StrictModel):
verdict: Literal["accept", "reject"]
step_by_step_check: str
blocking_issues: str = ""
patch_suggestion: str = ""
@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
class IterationRecord(StrictModel):
iteration: int
candidate_sha256: str
verdicts: list[JudgeVerdict]
unanimous: bool
pass_streak_after: int
class KernelRunResult(StrictModel):
item_id: str
status: Literal["accepted", "rejected"]
plan: KernelPlan
accepted_candidate: KernelCandidate | None = None
iterations: list[IterationRecord]
rejection_reason: str = ""
accepted_candidate_sha256: str | None = None
schema_version: str = SCHEMA_VERSION
@model_validator(mode="after")
def validate_terminal_state(self) -> "KernelRunResult":
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:
raise ValueError("rejected run must state a reason")
return self
class ModelCallRecord(StrictModel):
request_id: str
model: str
system_prompt: str
user_prompt: str
response_data: dict[str, Any]
raw_text: str
provider_response_id: str | None = None
usage: dict[str, Any] = Field(default_factory=dict)
created_at: str = Field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
class StageArtifact(StrictModel):
item_id: str
stage: str
payload_sha256: str
payload: dict[str, Any]
request_id: str | None = None
schema_version: str = SCHEMA_VERSION
created_at: str = Field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
|
