| """Strict proposal schema for language-model compiler integration.""" |
| from typing import Literal |
| from pydantic import Field, model_validator |
| from .schema import Record |
| from .compiled import Node,Plan |
|
|
| class Obligation(Record): |
| id: str |
| endpoint: str |
| unit: str |
| context_ids: list[str] |
| support: Literal['calibrated','proxy'] |
| threshold: float |
| direction: Literal['ge','le'] |
| scale: float=Field(gt=0) |
| threshold_source: str |
| worker: str |
|
|
| class Hypothesis(Record): |
| id: str |
| mechanism: str |
| evidence_ids: list[str] |
| opposing_evidence_ids: list[str] |
| counterexample: str |
| obligation_ids: list[str] |
|
|
| class PlanNode(Record): |
| id: str |
| tool: str |
| revision: str |
| inputs: list[str] |
| outputs: list[str] |
| cost: int=Field(ge=1) |
| attempts: int=Field(ge=1,le=3) |
|
|
| class PlanProposal(Record): |
| task_hash: str |
| evidence_snapshot_hash: str |
| hypotheses: list[Hypothesis]=Field(max_length=3) |
| obligations: list[Obligation] |
| nodes: list[PlanNode] |
| required: list[str]=Field(min_length=1) |
| budget: int=Field(ge=0) |
| seed: int |
| diagnostics: list[str] |
| @model_validator(mode='after') |
| def references(self): |
| ids={o.id for o in self.obligations} |
| if len(ids)!=len(self.obligations):raise ValueError('duplicate obligation') |
| for h in self.hypotheses: |
| if not h.evidence_ids or not set(h.obligation_ids)<=ids:raise ValueError('incomplete hypothesis') |
| return self |
| def execution_plan(self): |
| if self.diagnostics:raise ValueError('unresolved proposal diagnostics') |
| return Plan(tuple(Node(n.id,n.tool,n.revision,tuple(n.inputs),tuple(n.outputs),n.cost,n.attempts) for n in self.nodes),tuple(self.required),self.budget,self.seed) |
|
|
