chore: save current work
This commit is contained in:
@@ -13,7 +13,7 @@ import math
|
||||
from typing import Annotated, Any, Literal, TypeVar
|
||||
|
||||
from jsonschema import Draft202012Validator
|
||||
from pydantic import BaseModel, ConfigDict, Field, JsonValue, ValidationError, model_validator
|
||||
from pydantic import BaseModel, ConfigDict, Field, JsonValue, RootModel, ValidationError, model_validator
|
||||
|
||||
from app.cad_agent.domain.errors import ErrorCode, WorkflowError
|
||||
|
||||
@@ -31,84 +31,33 @@ class AcceptanceClaimInput(StrictDto):
|
||||
expected: dict[str, JsonValue] = Field(min_length=0, max_length=24)
|
||||
|
||||
|
||||
class RequirementInput(StrictDto):
|
||||
source_ids: list[Identifier] = Field(min_length=1, max_length=32)
|
||||
class SpecRequirementInput(StrictDto):
|
||||
statement: Annotated[str, Field(min_length=1, max_length=1000)]
|
||||
assumptions: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=16)
|
||||
acceptance_claims: list[AcceptanceClaimInput] = Field(min_length=1, max_length=16)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _source_ids_are_unique(self) -> "RequirementInput":
|
||||
if len(self.source_ids) != len(set(self.source_ids)):
|
||||
raise ValueError("source_ids must not contain duplicates")
|
||||
return self
|
||||
|
||||
class RequirementsSpec(StrictDto):
|
||||
outcome: Literal["ready"]
|
||||
summary: Annotated[str, Field(min_length=1, max_length=2000)]
|
||||
assumptions: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=32)
|
||||
requirements: list[SpecRequirementInput] = Field(min_length=1, max_length=32)
|
||||
|
||||
|
||||
class RequirementsDraftBatch(StrictDto):
|
||||
items: list[RequirementInput] = Field(min_length=1, max_length=8)
|
||||
class RequirementsClarification(StrictDto):
|
||||
outcome: Literal["clarification"]
|
||||
source_quotes: list[Annotated[str, Field(min_length=1, max_length=500)]] = Field(min_length=2, max_length=4)
|
||||
question: Annotated[str, Field(min_length=1, max_length=500)]
|
||||
|
||||
|
||||
class RequirementsPatch(StrictDto):
|
||||
target_draft_id: Identifier = Field(description="Current server-assigned draft ID to patch. This field belongs inside one patches[] entry.")
|
||||
op: Literal["replace", "remove"] = Field(description="replace supplies a complete replacement item; remove supplies a reason instead.")
|
||||
item: RequirementInput | None = Field(default=None, description="Complete replacement RequirementInput for op=replace. It must not include draft_id because the server preserves that ID.")
|
||||
reason: str | None = Field(default=None, min_length=1, max_length=360, description="Required only for op=remove; explains why the current draft item is removed.")
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _complete_patch(self) -> "RequirementsPatch":
|
||||
if self.op == "replace" and self.item is None:
|
||||
raise ValueError("replace requires a complete item")
|
||||
if self.op == "remove" and (self.item is not None or self.reason is None):
|
||||
raise ValueError("remove requires a reason and forbids item")
|
||||
return self
|
||||
|
||||
|
||||
class RequirementsPatchBatch(StrictDto):
|
||||
patches: list[RequirementsPatch] = Field(min_length=1, max_length=8, description="Patch entries. Example shape: {\"patches\":[{\"target_draft_id\":\"draft_001\",\"op\":\"replace\",\"item\":{...}}]}.")
|
||||
class RequirementsAuthorOutput(RootModel[Annotated[RequirementsSpec | RequirementsClarification, Field(discriminator="outcome")]]):
|
||||
pass
|
||||
|
||||
|
||||
class EmptyCommand(StrictDto):
|
||||
pass
|
||||
|
||||
|
||||
class ReviewNormalization(StrictDto):
|
||||
rule_id: Literal["full_circle_equal_spacing"]
|
||||
count: int = Field(ge=2, le=1024)
|
||||
declared_spacing_degrees: float = Field(gt=0, le=360)
|
||||
full_circle: bool
|
||||
|
||||
|
||||
class ReviewFinding(StrictDto):
|
||||
draft_id: Identifier
|
||||
source_ids: list[Identifier] = Field(min_length=1, max_length=32)
|
||||
finding_type: Literal[
|
||||
"missing_source_semantics",
|
||||
"claim_mismatch",
|
||||
"verification_gap",
|
||||
"derivable_conflict",
|
||||
"ambiguous_conflict",
|
||||
]
|
||||
description: Annotated[str, Field(min_length=1, max_length=1000)]
|
||||
question: str | None = Field(default=None, min_length=1, max_length=360)
|
||||
normalization: ReviewNormalization | None = None
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _finding_payload_matches_type(self) -> "ReviewFinding":
|
||||
if len(self.source_ids) != len(set(self.source_ids)):
|
||||
raise ValueError("source_ids must not contain duplicates")
|
||||
if self.finding_type == "ambiguous_conflict" and self.question is None:
|
||||
raise ValueError("ambiguous_conflict requires an answerable question")
|
||||
if self.finding_type == "derivable_conflict" and self.normalization is None:
|
||||
raise ValueError("derivable_conflict requires structured normalization data")
|
||||
if self.finding_type != "derivable_conflict" and self.normalization is not None:
|
||||
raise ValueError("normalization is allowed only for derivable_conflict")
|
||||
return self
|
||||
|
||||
|
||||
class RequirementsReview(StrictDto):
|
||||
findings: list[ReviewFinding] = Field(default_factory=list, max_length=128)
|
||||
|
||||
|
||||
class NextAction(StrictDto):
|
||||
working_head: Annotated[str, Field(pattern=r"^[a-z0-9_:-]{5,192}$")]
|
||||
intent: ShortText
|
||||
@@ -123,17 +72,18 @@ class NextAction(StrictDto):
|
||||
return self
|
||||
|
||||
|
||||
class StatelessNextAction(StrictDto):
|
||||
intent: ShortText
|
||||
operation: Identifier
|
||||
expected_change: ShortText
|
||||
|
||||
|
||||
class TopologyRequest(StrictDto):
|
||||
working_head: Annotated[str, Field(pattern=r"^[a-z0-9_:-]{5,192}$")]
|
||||
kind: Literal["face", "edge", "vertex", "plane", "axis", "body"] | None = None
|
||||
limit: int = Field(default=16, ge=1, le=64)
|
||||
|
||||
|
||||
class OperationContractRequest(StrictDto):
|
||||
working_head: Annotated[str, Field(pattern=r"^[a-z0-9_:-]{5,192}$")]
|
||||
atomic_id: Identifier
|
||||
|
||||
|
||||
class GeometryConclusion(StrictDto):
|
||||
working_head: Annotated[str, Field(pattern=r"^[a-z0-9_:-]{5,192}$")]
|
||||
evidence_refs: list[Identifier] = Field(min_length=1, max_length=16)
|
||||
@@ -154,6 +104,22 @@ class RollbackCheckpoint(StrictDto):
|
||||
reason: Annotated[str, Field(min_length=1, max_length=360)]
|
||||
|
||||
|
||||
class StatelessTopologyRequest(StrictDto):
|
||||
kind: Literal["face", "edge", "vertex", "plane", "axis", "body"] | None = None
|
||||
limit: int = Field(default=16, ge=1, le=64)
|
||||
|
||||
|
||||
class StatelessGeometryConclusion(StrictDto):
|
||||
root_cause: Annotated[str, Field(min_length=1, max_length=360)]
|
||||
decision: Literal["return_to_action_selection", "rollback"]
|
||||
corrective_intent: str | None = Field(default=None, min_length=1, max_length=360)
|
||||
|
||||
|
||||
class StatelessRollbackCheckpoint(StrictDto):
|
||||
checkpoint_token: Identifier
|
||||
reason: Annotated[str, Field(min_length=1, max_length=360)]
|
||||
|
||||
|
||||
class ClaimCoverage(StrictDto):
|
||||
claim_id: Identifier
|
||||
status: Literal["pass", "pending", "fail", "not_applicable"]
|
||||
@@ -175,6 +141,41 @@ class CandidateReview(StrictDto):
|
||||
issues: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=16)
|
||||
|
||||
|
||||
class StatelessCandidateReview(StrictDto):
|
||||
verdict: Literal["accept", "reject"]
|
||||
evidence: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=16)
|
||||
issues: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=16)
|
||||
|
||||
|
||||
class VisualClaimDecision(StrictDto):
|
||||
status: Literal["pass", "fail"]
|
||||
evidence: Annotated[str, Field(min_length=1, max_length=360)]
|
||||
|
||||
|
||||
class StatelessFinalReview(StrictDto):
|
||||
verdict: Literal["pass", "repair"]
|
||||
visual_claims: list[VisualClaimDecision] = Field(default_factory=list, max_length=128)
|
||||
evidence: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=16)
|
||||
issues: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=16)
|
||||
|
||||
|
||||
class ImageMeasurement(StrictDto):
|
||||
name: Annotated[str, Field(min_length=1, max_length=160)]
|
||||
value: float | None = None
|
||||
unit: Literal["mm", "degree", "count", "unknown"] = "unknown"
|
||||
evidence: Annotated[str, Field(min_length=1, max_length=360)]
|
||||
confidence: float = Field(ge=0, le=1)
|
||||
|
||||
|
||||
class ImageObservation(StrictDto):
|
||||
summary: Annotated[str, Field(min_length=1, max_length=2000)]
|
||||
visible_features: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=64)
|
||||
measurements: list[ImageMeasurement] = Field(default_factory=list, max_length=128)
|
||||
view_directions: list[Annotated[str, Field(min_length=1, max_length=120)]] = Field(default_factory=list, max_length=16)
|
||||
uncertainties: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=64)
|
||||
assumptions: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=64)
|
||||
|
||||
|
||||
class FinalReview(StrictDto):
|
||||
working_head: Annotated[str, Field(pattern=r"^[a-z0-9_:-]{5,192}$", description="Current server-issued working head from the final review facts.")]
|
||||
verdict: Literal["pass", "repair"] = Field(description="Required independent final decision. Set pass only when the supplied evidence supports every claim; otherwise set repair.")
|
||||
@@ -183,85 +184,39 @@ class FinalReview(StrictDto):
|
||||
issues: list[Annotated[str, Field(min_length=1, max_length=360)]] = Field(default_factory=list, max_length=16)
|
||||
|
||||
|
||||
def requirements_draft_schema(claim_one_of: dict[str, Any], source_ids: list[str]) -> dict[str, Any]:
|
||||
"""Bind claim and source enums for the current requirements snapshot."""
|
||||
schema = RequirementsDraftBatch.model_json_schema()
|
||||
requirement = schema.get("$defs", {}).get("RequirementInput")
|
||||
def requirements_spec_schema(claim_one_of: dict[str, Any]) -> dict[str, Any]:
|
||||
schema = RequirementsAuthorOutput.model_json_schema()
|
||||
requirement = schema.get("$defs", {}).get("SpecRequirementInput")
|
||||
if isinstance(requirement, dict):
|
||||
properties = requirement.get("properties", {})
|
||||
source_items = properties.get("source_ids", {}).get("items") if isinstance(properties, dict) and isinstance(properties.get("source_ids"), dict) else None
|
||||
if isinstance(source_items, dict):
|
||||
source_items["enum"] = source_ids
|
||||
claims = requirement.get("properties", {}).get("acceptance_claims")
|
||||
if isinstance(claims, dict):
|
||||
claims["items"] = deepcopy(claim_one_of)
|
||||
return schema
|
||||
|
||||
|
||||
def requirements_patch_schema(claim_one_of: dict[str, Any], source_ids: list[str], draft_ids: list[str]) -> dict[str, Any]:
|
||||
schema = RequirementsPatchBatch.model_json_schema()
|
||||
definitions = schema.get("$defs", {})
|
||||
requirement = definitions.get("RequirementInput") if isinstance(definitions, dict) else None
|
||||
if isinstance(requirement, dict):
|
||||
properties = requirement.get("properties", {})
|
||||
source_items = properties.get("source_ids", {}).get("items") if isinstance(properties, dict) and isinstance(properties.get("source_ids"), dict) else None
|
||||
if isinstance(source_items, dict):
|
||||
source_items["enum"] = source_ids
|
||||
claims = requirement.get("properties", {}).get("acceptance_claims")
|
||||
if isinstance(claims, dict):
|
||||
claims["items"] = deepcopy(claim_one_of)
|
||||
patch = definitions.get("RequirementsPatch") if isinstance(definitions, dict) else None
|
||||
if isinstance(patch, dict):
|
||||
target = patch.get("properties", {}).get("target_draft_id")
|
||||
if isinstance(target, dict):
|
||||
target["enum"] = draft_ids
|
||||
return schema
|
||||
|
||||
|
||||
def requirements_review_schema(source_ids: list[str], draft_ids: list[str]) -> dict[str, Any]:
|
||||
schema = RequirementsReview.model_json_schema()
|
||||
definitions = schema.get("$defs", {})
|
||||
finding = definitions.get("ReviewFinding") if isinstance(definitions, dict) else None
|
||||
if isinstance(finding, dict):
|
||||
properties = finding.get("properties", {})
|
||||
if isinstance(properties.get("draft_id"), dict):
|
||||
properties["draft_id"] = {"enum": draft_ids}
|
||||
source_items = properties.get("source_ids", {}).get("items") if isinstance(properties.get("source_ids"), dict) else None
|
||||
if isinstance(source_items, dict):
|
||||
source_items.clear()
|
||||
source_items.update({"enum": source_ids})
|
||||
return schema
|
||||
|
||||
|
||||
def next_action_schema(working_head: str, requirement_ids: list[str], atomic_ids: list[str]) -> dict[str, Any]:
|
||||
schema = NextAction.model_json_schema()
|
||||
def stateless_next_action_schema(atomic_ids: list[str]) -> dict[str, Any]:
|
||||
schema = StatelessNextAction.model_json_schema()
|
||||
properties = schema.get("properties", {})
|
||||
if isinstance(properties, dict):
|
||||
properties["working_head"] = {"const": working_head}
|
||||
if isinstance(properties.get("requirement_ids"), dict):
|
||||
properties["requirement_ids"]["items"] = {"enum": requirement_ids}
|
||||
properties["atomic_id"] = {"enum": atomic_ids}
|
||||
properties["operation"] = {"enum": atomic_ids}
|
||||
return schema
|
||||
|
||||
|
||||
def candidate_review_schema(candidate_id: str, working_head: str, claim_ids: list[str]) -> dict[str, Any]:
|
||||
"""Bind an independent candidate review to immutable candidate facts."""
|
||||
schema = CandidateReview.model_json_schema()
|
||||
def stateless_final_review_schema(visual_claim_count: int) -> dict[str, Any]:
|
||||
schema = StatelessFinalReview.model_json_schema()
|
||||
properties = schema.get("properties", {})
|
||||
if isinstance(properties, dict):
|
||||
properties["candidate_id"] = {"const": candidate_id}
|
||||
properties["working_head"] = {"const": working_head}
|
||||
_bind_claim_coverage_ids(schema, claim_ids)
|
||||
visual = properties.get("visual_claims") if isinstance(properties, dict) else None
|
||||
if isinstance(visual, dict):
|
||||
visual["minItems"] = visual_claim_count
|
||||
visual["maxItems"] = visual_claim_count
|
||||
return schema
|
||||
|
||||
|
||||
def final_review_schema(working_head: str, claim_ids: list[str]) -> dict[str, Any]:
|
||||
"""Bind final review output to the currently reviewable revision."""
|
||||
schema = FinalReview.model_json_schema()
|
||||
def stateless_rollback_checkpoint_schema(checkpoint_tokens: list[str]) -> dict[str, Any]:
|
||||
schema = StatelessRollbackCheckpoint.model_json_schema()
|
||||
properties = schema.get("properties", {})
|
||||
if isinstance(properties, dict):
|
||||
properties["working_head"] = {"const": working_head}
|
||||
_bind_claim_coverage_ids(schema, claim_ids)
|
||||
properties["checkpoint_token"] = {"enum": checkpoint_tokens}
|
||||
return schema
|
||||
|
||||
|
||||
@@ -273,15 +228,6 @@ def topology_request_schema(working_head: str) -> dict[str, Any]:
|
||||
return schema
|
||||
|
||||
|
||||
def operation_contract_request_schema(working_head: str, atomic_id: str) -> dict[str, Any]:
|
||||
schema = OperationContractRequest.model_json_schema()
|
||||
properties = schema.get("properties", {})
|
||||
if isinstance(properties, dict):
|
||||
properties["working_head"] = {"const": working_head}
|
||||
properties["atomic_id"] = {"const": atomic_id}
|
||||
return schema
|
||||
|
||||
|
||||
def geometry_conclusion_schema(working_head: str, evidence_refs: list[str]) -> dict[str, Any]:
|
||||
"""Bind a diagnostic conclusion to evidence generated for this head."""
|
||||
schema = GeometryConclusion.model_json_schema()
|
||||
|
||||
Reference in New Issue
Block a user