1118 lines
66 KiB
Python
1118 lines
66 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
from hashlib import sha256
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "backend"))
|
|
|
|
from app.cad_agent.adapters.artifact_store import FileArtifactStore
|
|
from app.cad_agent.adapters.event_publisher import IdempotentInProcessPublisher
|
|
from app.cad_agent.adapters.review_gateway import RenderedReviewGateway
|
|
from app.cad_agent.adapters.runtime import ProfileCadRuntime, RuntimeAdapterError
|
|
from app.cad_agent.adapters.sqlite_repository import SqliteTaskRepository
|
|
from app.cad_agent.adapters.verifier import RegistryVerifierExecutor
|
|
from app.cad_agent.application.capabilities import cached_model_capability, conformance_hash, conformance_tools, verify_model_capability
|
|
from app.cad_agent.application.action_handlers import ActionCommandHandler
|
|
from app.cad_agent.application.llm_contracts import (
|
|
AcceptanceClaimInput,
|
|
CandidateReview,
|
|
CompiledRequirementsSpec,
|
|
MarkdownDocument,
|
|
NextAction,
|
|
StatelessCandidateReview,
|
|
StatelessGeometryConclusion,
|
|
canonical_validate,
|
|
canonical_validate_schema,
|
|
compiled_requirements_schema,
|
|
sanitize_compiled_requirements_arguments,
|
|
stateless_final_review_schema,
|
|
stateless_next_action_schema,
|
|
stateless_rollback_checkpoint_schema,
|
|
)
|
|
from app.cad_agent.application.outbox import OutboxDispatcher
|
|
from app.cad_agent.application.requirements import RequirementsCommandHandler
|
|
from app.cad_agent.application.results import Accepted, Rejected, Waiting
|
|
from app.cad_agent.application.workflow import ModelIdentity, WorkflowConfig, WorkflowCoordinator
|
|
from app.cad_agent.domain.errors import ErrorCode, WorkflowError
|
|
from app.cad_agent.domain.operation_contract import fragment_schema, validate_fragment
|
|
from app.cad_agent.domain.state import PendingAction, TaskPhase, TaskState, legal_transitions, retry_resume_event, transition
|
|
from app.cad_agent.domain.verifier_registry import default_registry
|
|
from app.models.contracts import ChatMessage
|
|
from app.services.agent_service import AgentService
|
|
from app.services.library import CdslLibrary
|
|
from app.services.storage import WorkspaceStore
|
|
from app.settings import ProviderConfig, ProviderModel, Settings
|
|
|
|
|
|
def settings(root: Path) -> Settings:
|
|
author = ProviderConfig("author", "Author", "https://author.invalid/v1", "author-key", (ProviderModel("author-model"),))
|
|
reviewer = ProviderConfig("reviewer", "Reviewer", "https://reviewer.invalid/v1", "reviewer-key", (ProviderModel("reviewer-model", vision=True),))
|
|
return Settings(
|
|
task_root=root / "tasks",
|
|
conversation_root=root / "conversations",
|
|
library_root=ROOT / "backend" / "cdsl_library",
|
|
engine_root=ROOT / "backend" / "engine" / "cdsl_engine",
|
|
llm_base_url=author.base_url,
|
|
llm_api_key=author.api_key,
|
|
llm_model="author-model",
|
|
llm_timeout_s=1,
|
|
default_provider_id="author",
|
|
providers=(author, reviewer),
|
|
review_provider_id="reviewer",
|
|
review_model_id="reviewer-model",
|
|
)
|
|
|
|
|
|
def requirements_document() -> MarkdownDocument:
|
|
return MarkdownDocument(markdown="""# Design Understanding
|
|
|
|
Simple functional flange.
|
|
|
|
# Explicit User Requirements
|
|
|
|
- Create a simple flange.
|
|
|
|
# Engineering Defaults and Assumptions
|
|
|
|
- Use a circular body, central through bore, and four equally spaced mounting holes.
|
|
|
|
# Dimensions and Coordinate Convention
|
|
|
|
- Units are mm. The body is diameter 100 and thickness 10; bore diameter 30; four holes diameter 10 on radius 35.
|
|
|
|
# Open Uncertainties
|
|
|
|
- None.
|
|
""")
|
|
|
|
|
|
def completion_target() -> MarkdownDocument:
|
|
return MarkdownDocument(markdown="""# Completion Target
|
|
|
|
- [ ] One connected cylindrical flange body, 100 mm outer diameter and 10 mm thickness.
|
|
- [ ] Centered 30 mm through bore.
|
|
- [ ] Four 10 mm mounting holes on a circular pattern of 35 mm pitch radius.
|
|
""")
|
|
|
|
|
|
def compiled_flange() -> CompiledRequirementsSpec:
|
|
return CompiledRequirementsSpec.model_validate({"requirements": [
|
|
{"assumptions": [], "acceptance_claims": [{"claim_kind": "single_connected_body", "expected": {}}, {"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 100, "tolerance_mm": 0.1}}, {"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 10, "tolerance_mm": 0.1}}]},
|
|
{"assumptions": [], "acceptance_claims": [
|
|
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 30, "count": 1, "tolerance_mm": 0.1}},
|
|
{"claim_kind": "concentric_bore_to_outer_cylinder", "expected": {"bore_diameter_mm": 30, "outer_diameter_mm": 100, "tolerance_mm": 0.01}},
|
|
]},
|
|
{"assumptions": [], "acceptance_claims": [{"claim_kind": "circular_hole_pattern", "expected": {"count": 4, "diameter_mm": 10, "pitch_radius_mm": 35, "tolerance_mm": 0.1}}]},
|
|
]})
|
|
|
|
|
|
def modeling_plan() -> MarkdownDocument:
|
|
return MarkdownDocument(markdown="# Modeling Plan\n\n1. Create the circular flange body.\n2. Cut the centered bore.\n3. Add the circular mounting-hole pattern.\n")
|
|
|
|
|
|
def walk_keys(value: object) -> set[str]:
|
|
keys: set[str] = set()
|
|
if isinstance(value, dict):
|
|
keys.update(str(key) for key in value)
|
|
for item in value.values():
|
|
keys.update(walk_keys(item))
|
|
elif isinstance(value, list):
|
|
for item in value:
|
|
keys.update(walk_keys(item))
|
|
return keys
|
|
|
|
|
|
class CadV3ProtocolTests(unittest.TestCase):
|
|
def test_state_machine_has_no_requirements_review_phase(self) -> None:
|
|
self.assertNotIn("REVIEWING_REQUIREMENTS", {phase.value for phase in TaskPhase})
|
|
state = TaskState("cad_123456abcdef", TaskPhase.DRAFTING_REQUIREMENTS_DOCUMENT, 0)
|
|
document = transition(state, "requirements_document_written", requirements_document_path="requirements.md")
|
|
target = transition(document, "completion_target_written", completion_target_path="completion-target.md")
|
|
compiled = transition(target, "requirements_compiled", requirements_contract_path="requirements-contract.json")
|
|
approved = transition(compiled, "modeling_plan_written", modeling_plan_path="modeling-plan.md")
|
|
self.assertEqual(approved.phase, TaskPhase.AWAITING_ACTION)
|
|
self.assertNotIn("requirements_finalized", {event for _phase, event in legal_transitions()})
|
|
|
|
def test_waiting_retry_resumes_exact_source_phase(self) -> None:
|
|
state = TaskState("cad_123456abcdef", TaskPhase.DRAFTING_REQUIREMENTS_DOCUMENT, 0)
|
|
waiting = transition(state, "waiting_retry", error=ErrorCode.MODEL_PROTOCOL_CHECK_PENDING)
|
|
self.assertEqual(retry_resume_event(waiting), "resume_drafting_requirements_document")
|
|
resumed = transition(waiting, retry_resume_event(waiting) or "")
|
|
self.assertEqual(resumed.phase, TaskPhase.DRAFTING_REQUIREMENTS_DOCUMENT)
|
|
self.assertIsNone(resumed.retry_from_phase)
|
|
|
|
def test_llm_schemas_exclude_server_owned_runtime_ids(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
runtime = ProfileCadRuntime(settings(Path(temporary)))
|
|
schemas = [
|
|
MarkdownDocument.model_json_schema(),
|
|
compiled_requirements_schema(default_registry().expected_one_of_schema(exclude_claim_kinds=frozenset({"coaxial", "coplanar"})), 2),
|
|
stateless_next_action_schema(list(runtime.supported_atomic_ids())),
|
|
StatelessCandidateReview.model_json_schema(),
|
|
StatelessGeometryConclusion.model_json_schema(),
|
|
stateless_final_review_schema(2),
|
|
]
|
|
forbidden = {
|
|
"task_id", "working_head", "requirement_id", "requirement_ids", "claim_id",
|
|
"candidate_id", "action_id", "evidence_id", "evidence_refs", "source_id", "source_ids",
|
|
"draft_id", "attachment_id", "record_ids",
|
|
}
|
|
for schema in schemas:
|
|
self.assertFalse(walk_keys(schema) & forbidden, walk_keys(schema) & forbidden)
|
|
|
|
def test_compiled_requirements_ignores_non_executable_extra_fields(self) -> None:
|
|
schema = compiled_requirements_schema(default_registry().expected_one_of_schema(exclude_claim_kinds=frozenset({"coaxial", "coplanar"})), 1)
|
|
raw = json.dumps({
|
|
"assumptions": ["top-level notes from the compiler are not executable"],
|
|
"requirements": [{
|
|
"statement": "model-added copy of the checklist text",
|
|
"assumptions": [],
|
|
"acceptance_claims": [{
|
|
"claim_kind": "single_connected_body",
|
|
"expected": {},
|
|
"evidence": "not part of the compiler contract",
|
|
}],
|
|
}],
|
|
})
|
|
sanitized = sanitize_compiled_requirements_arguments(raw)
|
|
self.assertIsInstance(sanitized, str)
|
|
self.assertIsNone(canonical_validate_schema(sanitized, schema))
|
|
parsed = canonical_validate(sanitized, CompiledRequirementsSpec)
|
|
self.assertIsInstance(parsed, CompiledRequirementsSpec)
|
|
self.assertEqual(parsed.requirements[0].acceptance_claims[0].claim_kind, "single_connected_body")
|
|
|
|
def test_dynamic_tokens_are_enum_constrained(self) -> None:
|
|
rollback = stateless_rollback_checkpoint_schema(["checkpoint_one"])
|
|
self.assertEqual(rollback["properties"]["checkpoint_token"], {"enum": ["checkpoint_one"]})
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
runtime = ProfileCadRuntime(settings(Path(temporary)))
|
|
schema = fragment_schema(runtime.operation_contract("hole_blind"), selector_tokens=["selector_one"], reference_tokens=[])
|
|
selector = schema["properties"]["feature"]["properties"]["selector_tokens"]["items"]
|
|
self.assertEqual(selector, {"enum": ["selector_one"]})
|
|
|
|
def test_counterbore_can_reuse_a_matching_pilot_inside_an_annular_host_face(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
runtime = ProfileCadRuntime(settings(Path(temporary)))
|
|
host = {
|
|
"kind": "face",
|
|
"geometry": {
|
|
"surface_type": "plane", "center_mm": [0, 0, 34], "normal": [0, 0, 1],
|
|
"bbox_mm": [-48, -48, 34, 48, 48, 34],
|
|
"boundary_loops_mm": [
|
|
[[-48, -48, 34], [48, -48, 34], [48, 48, 34], [-48, 48, 34]],
|
|
[[-16, -16, 34], [16, -16, 34], [16, 16, 34], [-16, 16, 34]],
|
|
],
|
|
},
|
|
}
|
|
pilot = {
|
|
"kind": "face",
|
|
"geometry": {
|
|
"surface_type": "cylinder", "cylinder_role": "inner", "through": True,
|
|
"radius_mm": 16, "axis_origin_mm": [0, 0, 0], "axis_direction": [0, 0, 1],
|
|
"bbox_mm": [-16, -16, 0, 16, 16, 34],
|
|
},
|
|
}
|
|
fragment = {
|
|
"feature": {
|
|
"atomic_id": "hole_counterbore", "selector_tokens": ["host"],
|
|
"params": {"diameter_mm": 32, "depth_mm": 34, "counterbore_diameter_mm": 62, "counterbore_depth_mm": 12, "positions": [{"mm": [0, 0, 34]}]},
|
|
},
|
|
}
|
|
runtime._preflight_hole_positions_on_host_plane(fragment, {"host": host, "pilot": pilot}, None, False)
|
|
fragment["feature"]["params"]["diameter_mm"] = 30
|
|
with self.assertRaises(RuntimeAdapterError):
|
|
runtime._preflight_hole_positions_on_host_plane(fragment, {"host": host, "pilot": pilot}, None, False)
|
|
|
|
def test_counterbore_operation_verifier_measures_the_new_recess_not_the_existing_pilot(self) -> None:
|
|
class CounterboreVerifier:
|
|
def __init__(self) -> None:
|
|
self.operation_claims: list[dict[str, object]] = []
|
|
|
|
def evaluate(self, claims: list[dict[str, object]], _facts: dict[str, object]) -> list[dict[str, object]]:
|
|
if claims[0]["claim_id"] == "operation_parent_bore_count":
|
|
self.assertEqual(claims[0]["expected"]["diameter_mm"], 62.0)
|
|
return [{"evidence": {"actual_count": 0}}]
|
|
self.operation_claims = claims
|
|
return [{"claim_id": claim["claim_id"], "claim_kind": claim["claim_kind"], "deterministic": True, "status": "pass", "evidence": {}} for claim in claims]
|
|
|
|
def assertEqual(self, actual: object, expected: object) -> None:
|
|
if actual != expected:
|
|
raise AssertionError(f"{actual!r} != {expected!r}")
|
|
|
|
verifier = CounterboreVerifier()
|
|
handler = ActionCommandHandler(None, None, None, verifier)
|
|
action = PendingAction(
|
|
action_id="counterbore", working_head="cad_test:rev_001:v1", intent="Counterbore.", requirement_ids=(),
|
|
atomic_id="hole_counterbore", expected_change="Cut a counterbore.", contract_hash="contract", idempotency_key="key",
|
|
)
|
|
results = handler._operation_candidate_results(
|
|
action,
|
|
{"candidate_verifiers": ["cylindrical_bore"]},
|
|
{"features": [{"atomic_id": "hole_counterbore", "params": {"diameter_mm": 32, "counterbore_diameter_mm": 62, "positions": [{"mm": [0, 0, 34]}]}}]},
|
|
{"health": {}, "topology": {}, "report": {}},
|
|
parent_facts={"health": {}, "topology": {}, "report": {}},
|
|
require_through=False,
|
|
)
|
|
self.assertEqual(verifier.operation_claims[0]["expected"], {"diameter_mm": 62.0, "count": 1, "tolerance_mm": 0.01})
|
|
self.assertEqual(results[0]["status"], "pass")
|
|
|
|
def test_extrude_cut_rejects_a_slot_profile_floating_above_the_base(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
runtime = ProfileCadRuntime(settings(Path(temporary)))
|
|
base_face = {
|
|
"kind": "face",
|
|
"geometry": {
|
|
"surface_type": "plane", "center_mm": [0, 0, 16], "normal": [0, 0, 1],
|
|
"boundary_loops_mm": [[[-90, -50, 16], [90, -50, 16], [90, 50, 16], [-90, 50, 16]]],
|
|
},
|
|
}
|
|
boss_face = {
|
|
"kind": "face",
|
|
"geometry": {
|
|
"surface_type": "plane", "center_mm": [0, 0, 34], "normal": [0, 0, 1],
|
|
"boundary_loops_mm": [[[48, 0, 34], [0, 48, 34], [-48, 0, 34], [0, -48, 34]]],
|
|
},
|
|
}
|
|
fragment = {
|
|
"sketch": {
|
|
"workplane": {"origin_mm": [0, 0, 34], "normal": [0, 0, 1], "x_dir": [1, 0, 0]},
|
|
"profile": {"type": "analytic_contours", "contours": [{"closed": True, "role": "outer", "segments": [
|
|
{"type": "line", "start": [68, 40], "end": [85, 40]},
|
|
{"type": "line", "start": [85, 40], "end": [85, 49]},
|
|
{"type": "line", "start": [85, 49], "end": [68, 49]},
|
|
{"type": "line", "start": [68, 49], "end": [68, 40]},
|
|
]}]},
|
|
},
|
|
"feature": {"atomic_id": "extrude_cut_blind", "params": {"distance_mm": 2}},
|
|
}
|
|
with self.assertRaisesRegex(RuntimeAdapterError, "profile does not contact material"):
|
|
runtime._preflight_extrude_cut_contacts_material(fragment, {"base": base_face, "boss": boss_face})
|
|
fragment["sketch"]["workplane"]["origin_mm"][2] = 16
|
|
runtime._preflight_extrude_cut_contacts_material(fragment, {"base": base_face, "boss": boss_face})
|
|
|
|
def test_surface_attached_cut_direction_is_normalized_into_material(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
runtime = ProfileCadRuntime(settings(Path(temporary)))
|
|
base = {
|
|
"schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", "part_id": "cut_direction",
|
|
"geometry": {"sketches": [{
|
|
"id": "sketch_001",
|
|
"workplane": {"origin_mm": [0, 0, 0], "normal": [0, 0, 1], "x_dir": [1, 0, 0]},
|
|
"profile": {"type": "circle", "center": [0, 0], "radius_mm": 10},
|
|
}]},
|
|
"features": [{"id": "feature_001", "atomic_id": "extrude_add_blind", "params": {"distance_mm": 5}, "depends_on": [], "sketch_id": "sketch_001"}],
|
|
}
|
|
fragment = {
|
|
"sketch": {
|
|
"workplane": {"origin_mm": [0, 0, 5], "normal": [0, 0, 1], "x_dir": [1, 0, 0]},
|
|
"profile": {"type": "circle", "center": [6, 0], "radius_mm": 1},
|
|
},
|
|
"feature": {"atomic_id": "extrude_cut_blind", "params": {"distance_mm": 2}},
|
|
}
|
|
top_face = {
|
|
"kind": "face",
|
|
"geometry": {
|
|
"surface_type": "plane", "center_mm": [0, 0, 5], "normal": [0, 0, 1],
|
|
"boundary_loops_mm": [[[-10, -10, 5], [10, -10, 5], [10, 10, 5], [-10, 10, 5]]],
|
|
},
|
|
}
|
|
document, audit = runtime.materialize_fragment(
|
|
base,
|
|
fragment,
|
|
runtime.operation_contract("extrude_cut_blind"),
|
|
{"top": top_face},
|
|
runtime.reference_tokens(base),
|
|
)
|
|
self.assertTrue(document["features"][-1]["params"]["reverse"])
|
|
direction = next(item for item in audit["server_normalizations"] if item["path"] == "feature.params.reverse")
|
|
self.assertFalse(direction["submitted"])
|
|
self.assertTrue(direction["materialized"])
|
|
|
|
def test_sketch_workplane_candidates_prefer_the_broad_base_support(self) -> None:
|
|
candidates = WorkflowCoordinator._sketch_workplane_candidates({"records": [
|
|
{
|
|
"kind": "face",
|
|
"geometry": {
|
|
"surface_type": "plane", "center_mm": [0, 0, 34], "normal": [0, 0, 1],
|
|
"bbox_mm": [-48, -48, 34, 48, 48, 34],
|
|
},
|
|
},
|
|
{
|
|
"kind": "face",
|
|
"geometry": {
|
|
"surface_type": "plane", "center_mm": [0, 0, 16], "normal": [0, 0, 1],
|
|
"bbox_mm": [-90, -50, 16, 90, 50, 16],
|
|
},
|
|
},
|
|
{
|
|
"kind": "face",
|
|
"geometry": {
|
|
"surface_type": "plane", "center_mm": [90, 0, 8], "normal": [1, 0, 0],
|
|
"bbox_mm": [90, -50, 0, 90, 50, 16],
|
|
},
|
|
},
|
|
]})
|
|
self.assertEqual(candidates[0]["point_mm"], [0.0, 0.0, 16.0])
|
|
self.assertEqual(candidates[0]["footprint_bbox_area_mm2"], 18000.0)
|
|
self.assertEqual(candidates[1]["point_mm"], [0.0, 0.0, 34.0])
|
|
|
|
def test_replan_budget_spans_replacement_node_ids_at_one_checkpoint(self) -> None:
|
|
events = [
|
|
{
|
|
"event": "feature_node_failed", "node_id": node_id,
|
|
"atomic_id": "extrude_cut_blind", "checkpoint_revision": "rev_005", "terminal": True,
|
|
}
|
|
for node_id in ("corner_slots_v1", "corner_slots_v2", "corner_slots_v3")
|
|
]
|
|
|
|
class Repository:
|
|
@staticmethod
|
|
def ledger_events(_task_id: str) -> list[dict[str, object]]:
|
|
return events
|
|
|
|
workflow = object.__new__(WorkflowCoordinator)
|
|
workflow.repository = Repository()
|
|
state = TaskState("cad_123456abcdef", TaskPhase.REPLANNING_FEATURE_SUBGRAPH, 10, active_revision="rev_005")
|
|
exhausted = workflow._feature_replan_exhausted(state.task_id, state)
|
|
self.assertIsNotNone(exhausted)
|
|
self.assertEqual(exhausted["terminal_failure_count"], 3)
|
|
self.assertEqual(exhausted["node_ids"], ["corner_slots_v1", "corner_slots_v2", "corner_slots_v3"])
|
|
|
|
def test_root_extrusion_schema_fixes_world_xy_datum_without_deciding_z(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
runtime = ProfileCadRuntime(settings(Path(temporary)))
|
|
contract = runtime.operation_contract("extrude_add_blind")
|
|
fragment = {
|
|
"sketch": {"workplane": {"origin_mm": [0, -6, 0], "normal": [0, 0, 1], "x_dir": [1, 0, 0]}, "profile": {"type": "circle", "center": [0, 0], "radius_mm": 60}},
|
|
"feature": {"atomic_id": "extrude_add_blind", "params": {"distance_mm": 12}},
|
|
}
|
|
self.assertTrue(validate_fragment(contract, fragment, selector_tokens=[], root_xy_datum=True))
|
|
fragment["sketch"]["workplane"]["origin_mm"] = [0, 0, -6]
|
|
self.assertEqual(validate_fragment(contract, fragment, selector_tokens=[], root_xy_datum=True), [])
|
|
|
|
def test_runtime_keeps_executable_feature_prefix_when_later_feature_is_invalid(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
runtime = ProfileCadRuntime(settings(Path(temporary)))
|
|
cdsl = {
|
|
"schema": "cad.cdsl.llm.v1",
|
|
"schema_version": "1.1.0",
|
|
"kind": "part",
|
|
"part_id": "partial_rebuild",
|
|
"geometry": {"sketches": [{
|
|
"id": "sketch_001",
|
|
"workplane": {"origin_mm": [0, 0, 0], "normal": [0, 0, 1], "x_dir": [1, 0, 0]},
|
|
"profile": {"type": "circle", "center": [0, 0], "radius_mm": 10},
|
|
}]},
|
|
"features": [
|
|
{"id": "feature_001", "atomic_id": "extrude_add_blind", "params": {"distance_mm": 5}, "depends_on": [], "sketch_id": "sketch_001"},
|
|
{"id": "feature_002", "atomic_id": "not_an_engine_operation", "params": {}, "depends_on": ["feature_001"]},
|
|
],
|
|
}
|
|
rebuilt, failures = runtime.rebuild_best_effort(cdsl, str(Path(temporary) / "candidate"), "partial_rebuild", "candidate")
|
|
self.assertEqual(rebuilt["executed_feature_ids"], ["feature_001"])
|
|
self.assertEqual(len(failures), 1)
|
|
self.assertEqual(failures[0]["feature_id"], "feature_002")
|
|
|
|
def test_action_submission_keeps_partial_feature_batch_for_review(self) -> None:
|
|
class PassingVerifier:
|
|
def evaluate(self, claims: list[dict[str, object]], _facts: dict[str, object]) -> list[dict[str, object]]:
|
|
return [
|
|
{
|
|
"claim_id": str(claim.get("claim_id") or ""),
|
|
"claim_kind": str(claim.get("claim_kind") or ""),
|
|
"deterministic": True,
|
|
"status": "pass",
|
|
"evidence": {},
|
|
}
|
|
for claim in claims
|
|
]
|
|
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
runtime = ProfileCadRuntime(settings(root))
|
|
requirements = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
actions = ActionCommandHandler(repository, artifacts, runtime, PassingVerifier())
|
|
task_id = "cad_123456abcdef"
|
|
repository.create_task(task_id, "Create a flange.")
|
|
artifacts.initialize_task(task_id, "Create a flange.")
|
|
requirements.submit_requirements_document(task_id, requirements_document(), invocation_id="requirements_document")
|
|
requirements.submit_completion_target(task_id, completion_target(), invocation_id="completion_target")
|
|
requirements.submit_compiled_spec(task_id, compiled_flange(), invocation_id="requirements_compile")
|
|
requirements.submit_modeling_plan(task_id, modeling_plan(), invocation_id="modeling_plan")
|
|
state = repository.get_state(task_id)
|
|
proposal = NextAction(
|
|
working_head=state.working_head,
|
|
intent="Create a two-feature batch.",
|
|
requirement_ids=["req_001"],
|
|
atomic_id="extrude_add_blind",
|
|
expected_change="Keep the executable part of the batch.",
|
|
)
|
|
self.assertIsInstance(actions.propose_next_action(task_id, proposal, invocation_id="action"), Accepted)
|
|
fragment = {
|
|
"sketch": {
|
|
"workplane": {"origin_mm": [0, 0, 0], "normal": [0, 0, 1], "x_dir": [1, 0, 0]},
|
|
"profile": {"type": "circle", "center": [0, 0], "radius_mm": 10},
|
|
},
|
|
"feature": {"atomic_id": "extrude_add_blind", "params": {"distance_mm": 5}},
|
|
}
|
|
cdsl = {
|
|
"schema": "cad.cdsl.llm.v1",
|
|
"schema_version": "1.1.0",
|
|
"kind": "part",
|
|
"part_id": "partial_batch",
|
|
"geometry": {"sketches": []},
|
|
"features": [
|
|
{"id": "feature_001", "atomic_id": "extrude_add_blind", "params": {"distance_mm": 5}, "depends_on": []},
|
|
{"id": "feature_002", "atomic_id": "extrude_add_blind", "params": {"distance_mm": 5}, "depends_on": ["feature_001"]},
|
|
],
|
|
}
|
|
audit = {
|
|
"schema_version": "cad.v3.fragment-audit.v1",
|
|
"atomic_id": "extrude_add_blind",
|
|
"fragment_hash": "hash",
|
|
"contract_hash": runtime.operation_contract("extrude_add_blind")["contract_hash"],
|
|
"assigned_feature_ids": ["feature_001", "feature_002"],
|
|
"assigned_sketch_ids": [],
|
|
"selector_snapshot_id": "",
|
|
"selector_tokens": [],
|
|
"reference_snapshot_id": "",
|
|
"reference_tokens": [],
|
|
}
|
|
rebuilt = {
|
|
"executed_feature_ids": ["feature_001"],
|
|
"health": {"solid_count": 1},
|
|
"topology": {"records": []},
|
|
"report": {},
|
|
"render_manifest": {},
|
|
"paths": {"cdsl": "model.cdsl.json", "step": "model.step", "glb": "model.glb", "topology": "model.topology.json", "report": "rebuild-report.json"},
|
|
}
|
|
operation_failures = [{"feature_index": 1, "feature_id": "feature_002", "message": "failed after feature_001"}]
|
|
with patch.object(runtime, "materialize_fragment", return_value=(cdsl, audit)), patch.object(runtime, "rebuild_best_effort", return_value=(rebuilt, operation_failures)):
|
|
result = actions.submit_cdsl_fragment(task_id, fragment, invocation_id="fragment")
|
|
self.assertIsInstance(result, Accepted)
|
|
reviewing = repository.get_state(task_id)
|
|
self.assertEqual(reviewing.phase, TaskPhase.CANDIDATE_REVIEW)
|
|
candidate = artifacts.read_stage_json(task_id, reviewing.candidate_stage_id, "candidate.json") or {}
|
|
self.assertEqual(candidate["executed_feature_ids"], ["feature_001"])
|
|
self.assertEqual(candidate["operation_failures"], operation_failures)
|
|
|
|
def test_best_effort_transition_completes_from_an_executable_checkpoint(self) -> None:
|
|
state = TaskState("cad_123456abcdef", TaskPhase.AWAITING_ACTION, 7, active_revision="rev_001", repair_required=True)
|
|
completed = transition(state, "best_effort_completed", error=ErrorCode.BEST_EFFORT_COMPLETED, repair_required=False)
|
|
self.assertEqual(completed.phase, TaskPhase.COMPLETED)
|
|
self.assertFalse(completed.repair_required)
|
|
|
|
def test_review_rejection_publishes_the_executable_checkpoint_for_repair(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
runtime = ProfileCadRuntime(settings(root))
|
|
requirements = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
actions = ActionCommandHandler(repository, artifacts, runtime, RegistryVerifierExecutor(default_registry()))
|
|
task_id = "cad_123456abcdef"
|
|
repository.create_task(task_id, "Create a simple flange.")
|
|
artifacts.initialize_task(task_id, "Create a simple flange.")
|
|
requirements.submit_requirements_document(task_id, requirements_document(), invocation_id="requirements_document")
|
|
requirements.submit_completion_target(task_id, completion_target(), invocation_id="completion_target")
|
|
requirements.submit_compiled_spec(task_id, compiled_flange(), invocation_id="requirements_compile")
|
|
requirements.submit_modeling_plan(task_id, modeling_plan(), invocation_id="modeling_plan")
|
|
state = repository.get_state(task_id)
|
|
proposal = NextAction(
|
|
working_head=state.working_head,
|
|
intent="Create the base body.",
|
|
requirement_ids=["req_001"],
|
|
atomic_id="extrude_add_blind",
|
|
expected_change="Create the circular body.",
|
|
)
|
|
self.assertIsInstance(actions.propose_next_action(task_id, proposal, invocation_id="action"), Accepted)
|
|
fragment = {
|
|
"sketch": {
|
|
"workplane": {"origin_mm": [0, 0, 0], "normal": [0, 0, 1], "x_dir": [1, 0, 0]},
|
|
"profile": {"type": "circle", "center": [0, 0], "radius_mm": 50},
|
|
},
|
|
"feature": {"atomic_id": "extrude_add_blind", "params": {"distance_mm": 10}},
|
|
}
|
|
self.assertIsInstance(actions.submit_cdsl_fragment(task_id, fragment, invocation_id="fragment"), Accepted)
|
|
reviewing = repository.get_state(task_id)
|
|
candidate = artifacts.read_stage_json(task_id, reviewing.candidate_stage_id, "candidate.json") or {}
|
|
review = CandidateReview(
|
|
candidate_id=reviewing.candidate_id,
|
|
working_head=reviewing.pending_action.working_head,
|
|
verdict="reject",
|
|
claim_coverage=[
|
|
{"claim_id": str(item["claim_id"]), "status": str(item["status"]), "evidence_refs": []}
|
|
for item in candidate["claim_results"]
|
|
],
|
|
evidence=["Base body is executable."],
|
|
issues=["The bore and bolt holes remain to be added."],
|
|
)
|
|
result = actions.record_candidate_review(task_id, review, invocation_id="review")
|
|
self.assertIsInstance(result, Accepted)
|
|
self.assertEqual(result.payload["status"], "accepted_with_issues")
|
|
published = repository.get_state(task_id)
|
|
self.assertEqual(published.phase, TaskPhase.AWAITING_ACTION)
|
|
self.assertEqual(published.active_revision, "rev_001")
|
|
self.assertTrue(published.repair_required)
|
|
completed = actions.finalize_best_effort(task_id, reason=ErrorCode.NO_PROGRESS_LIMIT, invocation_id="best_effort")
|
|
self.assertIsInstance(completed, Accepted)
|
|
self.assertEqual(repository.get_state(task_id).phase, TaskPhase.COMPLETED)
|
|
|
|
def test_root_checkpoint_is_not_offered_as_a_rollback_target(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
configured = settings(root)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
runtime = ProfileCadRuntime(configured)
|
|
initial = repository.create_task("cad_123456abcdef", "Create a flange.")
|
|
document = transition(initial, "requirements_document_written", requirements_document_path="requirements.md")
|
|
target = transition(document, "completion_target_written", completion_target_path="completion-target.md")
|
|
compiled = transition(target, "requirements_compiled", requirements_contract_path="requirements-contract.json")
|
|
awaiting = transition(compiled, "modeling_plan_written", modeling_plan_path="modeling-plan.md", repair_required=True)
|
|
self.assertTrue(repository.compare_and_swap(document))
|
|
self.assertTrue(repository.compare_and_swap(target))
|
|
self.assertTrue(repository.compare_and_swap(compiled))
|
|
self.assertTrue(repository.compare_and_swap(awaiting, events=[{
|
|
"event": "geometry_conclusion",
|
|
"decision": "rollback",
|
|
"working_head": awaiting.working_head,
|
|
}]))
|
|
actions = ActionCommandHandler(repository, artifacts, runtime, default_registry())
|
|
self.assertFalse(actions.rollback_available(awaiting.task_id))
|
|
|
|
def test_geometry_conclusion_is_stateless_for_the_author(self) -> None:
|
|
schema = StatelessGeometryConclusion.model_json_schema()
|
|
self.assertFalse({"working_head", "evidence_refs"} & walk_keys(schema))
|
|
|
|
def test_outer_cylinder_span_merges_oppositely_oriented_two_sided_faces(self) -> None:
|
|
def outer_face(record_id: str, direction: list[float], bbox: list[float]) -> dict[str, object]:
|
|
return {
|
|
"record_id": record_id,
|
|
"geometry": {
|
|
"surface_type": "cylinder",
|
|
"cylinder_role": "outer",
|
|
"radius_mm": 60.0,
|
|
"axis_origin_mm": [0.0, 0.0, 0.0],
|
|
"axis_direction": direction,
|
|
"bbox_mm": bbox,
|
|
},
|
|
}
|
|
|
|
facts = {"topology": {"records": [
|
|
outer_face("upper", [0.0, 0.0, -1.0], [-60.0, -60.0, 0.0, 60.0, 60.0, 6.0]),
|
|
outer_face("lower", [0.0, 0.0, 1.0], [-60.0, -60.0, -6.0, 60.0, 60.0, 0.0]),
|
|
]}}
|
|
result = default_registry().evaluate(
|
|
"outer_cylindrical_surface",
|
|
{"diameter_mm": 120.0, "count": 1, "axial_span_mm": 12.0},
|
|
facts,
|
|
)
|
|
self.assertEqual(result["status"], "pass")
|
|
self.assertEqual(result["evidence"]["axial_spans_mm"], [12.0])
|
|
self.assertEqual(result["evidence"]["tolerance_mm"], 0.1)
|
|
|
|
def test_compiler_persists_default_tolerance_for_axial_outer_cylinder(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
handler = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
task_id = "cad_123456abcdef"
|
|
repository.create_task(task_id, "Create a flange.")
|
|
artifacts.initialize_task(task_id, "Create a flange.")
|
|
handler.submit_requirements_document(task_id, requirements_document(), invocation_id="requirements_document")
|
|
handler.submit_completion_target(task_id, completion_target(), invocation_id="completion_target")
|
|
compiled = CompiledRequirementsSpec.model_validate({"requirements": [
|
|
{"assumptions": [], "acceptance_claims": [{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 100, "axial_span_mm": 10, "count": 1}}]},
|
|
{"assumptions": [], "acceptance_claims": [
|
|
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 30, "count": 1, "tolerance_mm": 0.1}},
|
|
{"claim_kind": "concentric_bore_to_outer_cylinder", "expected": {"bore_diameter_mm": 30, "outer_diameter_mm": 100, "tolerance_mm": 0.01}},
|
|
]},
|
|
{"assumptions": [], "acceptance_claims": [{"claim_kind": "circular_hole_pattern", "expected": {"diameter_mm": 10, "count": 4, "pitch_radius_mm": 35, "tolerance_mm": 0.1}}]},
|
|
]})
|
|
self.assertIsInstance(handler.submit_compiled_spec(task_id, compiled, invocation_id="requirements_compile"), Accepted)
|
|
state = repository.get_state(task_id)
|
|
contract = artifacts.read_requirements_contract(task_id, state.requirements_contract_path) or {}
|
|
first_claim = contract["requirements"][0]["acceptance_claims"][0]
|
|
self.assertEqual(first_claim["expected"]["tolerance_mm"], 0.1)
|
|
|
|
def test_record_bound_compiler_claims_are_visualized_before_contract_freeze(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
handler = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
task_id = "cad_123456abcdef"
|
|
repository.create_task(task_id, "Create a simple flange.")
|
|
artifacts.initialize_task(task_id, "Create a simple flange.")
|
|
handler.submit_requirements_document(task_id, requirements_document(), invocation_id="requirements_document")
|
|
handler.submit_completion_target(task_id, completion_target(), invocation_id="completion_target")
|
|
compiled = CompiledRequirementsSpec.model_validate({"requirements": [
|
|
{"assumptions": [], "acceptance_claims": [{"claim_kind": "coaxial", "expected": {"record_ids": ["outer", "bore"], "tolerance": 0.01}}]},
|
|
{"assumptions": [], "acceptance_claims": [
|
|
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 30, "count": 1, "tolerance_mm": 0.1}},
|
|
{"claim_kind": "concentric_bore_to_outer_cylinder", "expected": {"bore_diameter_mm": 30, "outer_diameter_mm": 100, "tolerance_mm": 0.01}},
|
|
]},
|
|
{"assumptions": [], "acceptance_claims": [{"claim_kind": "coplanar", "expected": {"record_ids": ["top_face", "bottom_face"], "tolerance_mm": 0.1}}]},
|
|
]})
|
|
self.assertIsInstance(handler.submit_compiled_spec(task_id, compiled, invocation_id="requirements_compile"), Accepted)
|
|
state = repository.get_state(task_id)
|
|
contract = artifacts.read_requirements_contract(task_id, state.requirements_contract_path) or {}
|
|
claims = [claim for requirement in contract["requirements"] for claim in requirement["acceptance_claims"]]
|
|
self.assertEqual([claim["claim_kind"] for claim in claims], ["visual", "through_cylindrical_bore", "concentric_bore_to_outer_cylinder", "visual"])
|
|
self.assertEqual([claim["verification_mode"] for claim in claims], ["visual", "deterministic", "deterministic", "visual"])
|
|
self.assertTrue(any("coaxial verifier" in warning for warning in contract["verification_warnings"]))
|
|
self.assertTrue(any("coplanar verifier" in warning for warning in contract["verification_warnings"]))
|
|
|
|
def test_unbacked_coaxial_bore_group_is_not_frozen_as_a_deterministic_claim(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
handler = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
output = CompiledRequirementsSpec.model_validate({"requirements": [
|
|
{"assumptions": [], "acceptance_claims": [
|
|
{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 40, "count": 1, "tolerance_mm": 0.01}},
|
|
]},
|
|
{"assumptions": [], "acceptance_claims": [
|
|
{"claim_kind": "coaxial_through_bore_group", "expected": {"diameter_mm": 40, "count": 2, "tolerance_mm": 0.01}},
|
|
]},
|
|
]})
|
|
normalized, warnings = handler._normalize_compiled_spec(output, ["A through bore.", "The bore is concentric with the outer profile."])
|
|
self.assertEqual(normalized.requirements[1].acceptance_claims[0].claim_kind, "visual")
|
|
self.assertTrue(any("no matching multi-bore target" in warning for warning in warnings))
|
|
|
|
def test_obround_slot_is_not_compiled_as_a_corner_bore_pattern(self) -> None:
|
|
output = CompiledRequirementsSpec.model_validate({"requirements": [{
|
|
"assumptions": [],
|
|
"acceptance_claims": [{
|
|
"claim_kind": "rectangular_corner_through_bore_pattern",
|
|
"expected": {"diameter_mm": 9, "count": 4, "edge_offset_mm": 26, "tolerance_mm": 0.1},
|
|
}],
|
|
}]})
|
|
normalized, warnings = RequirementsCommandHandler(None, None, default_registry())._normalize_compiled_spec(
|
|
output,
|
|
["Four 26 x 9 mm oblong adjustment slots are present at the four corners."],
|
|
)
|
|
claim = normalized.requirements[0].acceptance_claims[0]
|
|
self.assertEqual(claim.claim_kind, "visual")
|
|
self.assertEqual(claim.expected["description"], "Four 26 x 9 mm oblong adjustment slots are present at the four corners.")
|
|
self.assertTrue(any("describes an obround slot" in warning for warning in warnings))
|
|
|
|
def test_centered_bore_checklist_item_requires_concentric_claim_coverage(self) -> None:
|
|
output = CompiledRequirementsSpec.model_validate({"requirements": [{
|
|
"assumptions": [],
|
|
"acceptance_claims": [{"claim_kind": "through_cylindrical_bore", "expected": {"diameter_mm": 40, "count": 1, "tolerance_mm": 0.01}}],
|
|
}]})
|
|
errors = RequirementsCommandHandler._relationship_claim_errors(
|
|
output,
|
|
["A centered 40 mm through bore is present."],
|
|
)
|
|
self.assertEqual(errors[0]["path"], "/requirements/0/acceptance_claims")
|
|
output.requirements[0].acceptance_claims.append(AcceptanceClaimInput.model_validate({
|
|
"claim_kind": "concentric_bore_to_outer_cylinder",
|
|
"expected": {"bore_diameter_mm": 40, "outer_diameter_mm": 120, "tolerance_mm": 0.01},
|
|
}))
|
|
self.assertEqual(
|
|
RequirementsCommandHandler._relationship_claim_errors(output, ["A centered 40 mm through bore is present."]),
|
|
[],
|
|
)
|
|
|
|
def test_compiler_derives_concentric_claim_from_frozen_outer_cylinder_and_centered_bore(self) -> None:
|
|
output = CompiledRequirementsSpec.model_validate({"requirements": [
|
|
{"assumptions": [], "acceptance_claims": [{
|
|
"claim_kind": "outer_cylindrical_surface",
|
|
"expected": {"diameter_mm": 120, "axial_span_mm": 12, "count": 1, "tolerance_mm": 0.01},
|
|
}]},
|
|
{"assumptions": [], "acceptance_claims": [{
|
|
"claim_kind": "through_cylindrical_bore",
|
|
"expected": {"diameter_mm": 40, "count": 1, "tolerance_mm": 0.01},
|
|
}]},
|
|
]})
|
|
normalized, _ = RequirementsCommandHandler(None, None, default_registry())._normalize_compiled_spec(
|
|
output,
|
|
["A 120 mm cylindrical outer flange is present.", "A centered 40 mm through bore is present."],
|
|
)
|
|
derived = normalized.requirements[1].acceptance_claims[-1]
|
|
self.assertEqual(derived.claim_kind, "concentric_bore_to_outer_cylinder")
|
|
self.assertEqual(derived.expected, {"bore_diameter_mm": 40.0, "outer_diameter_mm": 120.0, "tolerance_mm": 0.01})
|
|
self.assertEqual(
|
|
RequirementsCommandHandler._relationship_claim_errors(
|
|
normalized,
|
|
["A 120 mm cylindrical outer flange is present.", "A centered 40 mm through bore is present."],
|
|
),
|
|
[],
|
|
)
|
|
|
|
def test_concentric_bore_to_outer_cylinder_verifier_measures_axis_offset(self) -> None:
|
|
outer = {
|
|
"record_id": "outer", "geometry": {
|
|
"surface_type": "cylinder", "cylinder_role": "outer", "radius_mm": 60.0,
|
|
"axis_origin_mm": [0.0, 0.0, 0.0], "axis_direction": [0.0, 0.0, 1.0],
|
|
"bbox_mm": [-60.0, -60.0, 0.0, 60.0, 60.0, 12.0],
|
|
},
|
|
}
|
|
bore = {
|
|
"record_id": "bore", "geometry": {
|
|
"surface_type": "cylinder", "cylinder_role": "inner", "radius_mm": 20.0,
|
|
"axis_origin_mm": [0.0, 0.0, 0.0], "axis_direction": [0.0, 0.0, 1.0],
|
|
"bbox_mm": [-20.0, -20.0, 0.0, 20.0, 20.0, 12.0], "through": True,
|
|
},
|
|
}
|
|
expected = {"bore_diameter_mm": 40.0, "outer_diameter_mm": 120.0, "tolerance_mm": 0.01}
|
|
registry = default_registry()
|
|
result = registry.evaluate("concentric_bore_to_outer_cylinder", expected, {"topology": {"records": [outer, bore]}})
|
|
self.assertEqual(result["status"], "pass")
|
|
bore["geometry"]["axis_origin_mm"] = [0.1, 0.0, 0.0]
|
|
result = registry.evaluate("concentric_bore_to_outer_cylinder", expected, {"topology": {"records": [outer, bore]}})
|
|
self.assertEqual(result["status"], "fail")
|
|
self.assertAlmostEqual(result["evidence"]["axis_distance_mm"], 0.1)
|
|
|
|
def test_local_cylindrical_span_does_not_become_global_bbox_requirement(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
handler = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
task_id = "cad_123456abcdef"
|
|
repository.create_task(task_id, "Create a stepped hub adapter.")
|
|
artifacts.initialize_task(task_id, "Create a stepped hub adapter.")
|
|
handler.submit_requirements_document(task_id, requirements_document(), invocation_id="requirements_document")
|
|
handler.submit_completion_target(task_id, MarkdownDocument(markdown="""# Completion Target
|
|
|
|
- [ ] A centered solid cylindrical flange body is present with 120 mm outer diameter and 12 mm thickness.
|
|
"""), invocation_id="completion_target")
|
|
compiled = CompiledRequirementsSpec.model_validate({"requirements": [{
|
|
"assumptions": [],
|
|
"acceptance_claims": [
|
|
{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 120, "count": 1, "tolerance_mm": 0.1}},
|
|
{"claim_kind": "bbox_dimension_mm", "expected": {"axis": "z", "value": 12, "tolerance_mm": 0.1}},
|
|
],
|
|
}]})
|
|
self.assertIsInstance(handler.submit_compiled_spec(task_id, compiled, invocation_id="requirements_compile"), Accepted)
|
|
state = repository.get_state(task_id)
|
|
contract = artifacts.read_requirements_contract(task_id, state.requirements_contract_path) or {}
|
|
claims = contract["requirements"][0]["acceptance_claims"]
|
|
self.assertEqual([claim["claim_kind"] for claim in claims], ["outer_cylindrical_surface"])
|
|
self.assertEqual(claims[0]["expected"]["axial_span_mm"], 12)
|
|
self.assertTrue(any("Global bbox Z verifier" in warning for warning in contract["verification_warnings"]))
|
|
|
|
def test_requirements_markdown_is_not_rejected_for_missing_headings(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
handler = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
task_id = "cad_123456abcdef"
|
|
repository.create_task(task_id, "Create a flange.")
|
|
artifacts.initialize_task(task_id, "Create a flange.")
|
|
result = handler.submit_requirements_document(task_id, MarkdownDocument(markdown="A simple circular flange with a bore."), invocation_id="plain_markdown")
|
|
self.assertIsInstance(result, Accepted)
|
|
self.assertEqual(repository.get_state(task_id).phase, TaskPhase.DRAFTING_COMPLETION_TARGET)
|
|
|
|
def test_server_bound_action_accepts_more_than_five_checklist_targets(self) -> None:
|
|
action = NextAction(
|
|
working_head="cad_123456abcdef:root:v4",
|
|
intent="Create the flange body.",
|
|
requirement_ids=[f"req_{position:03d}" for position in range(1, 8)],
|
|
atomic_id="extrude_add_blind",
|
|
expected_change="Add the first solid body.",
|
|
)
|
|
self.assertEqual(len(action.requirement_ids), 7)
|
|
|
|
def test_markdown_documents_freeze_before_compiled_flange_contract(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
handler = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
task_id = "cad_123456abcdef"
|
|
repository.create_task(task_id, "Create a simple flange.")
|
|
artifacts.initialize_task(task_id, "Create a simple flange.")
|
|
self.assertIsInstance(handler.submit_requirements_document(task_id, requirements_document(), invocation_id="requirements_document"), Accepted)
|
|
self.assertIsInstance(handler.submit_completion_target(task_id, completion_target(), invocation_id="completion_target"), Accepted)
|
|
self.assertIsInstance(handler.submit_compiled_spec(task_id, compiled_flange(), invocation_id="requirements_compile"), Accepted)
|
|
self.assertIsInstance(handler.submit_modeling_plan(task_id, modeling_plan(), invocation_id="modeling_plan"), Accepted)
|
|
state = repository.get_state(task_id)
|
|
self.assertEqual(state.phase, TaskPhase.AWAITING_ACTION)
|
|
contract = artifacts.read_requirements_contract(task_id, state.requirements_contract_path) or {}
|
|
self.assertEqual(len(contract["requirements"]), 3)
|
|
claim_kinds = {claim["claim_kind"] for item in contract["requirements"] for claim in item["acceptance_claims"]}
|
|
self.assertTrue({"single_connected_body", "through_cylindrical_bore", "circular_hole_pattern"}.issubset(claim_kinds))
|
|
self.assertTrue((artifacts.task_dir(task_id) / "requirements.md").is_file())
|
|
self.assertTrue((artifacts.task_dir(task_id) / "completion-target.md").is_file())
|
|
self.assertTrue((artifacts.task_dir(task_id) / "modeling-plan.md").is_file())
|
|
|
|
def test_invalid_verifier_contract_is_rejected_without_state_change(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
handler = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
task_id = "cad_123456abcdef"
|
|
repository.create_task(task_id, "Create one solid.")
|
|
artifacts.initialize_task(task_id, "Create one solid.")
|
|
handler.submit_requirements_document(task_id, requirements_document(), invocation_id="requirements_document")
|
|
handler.submit_completion_target(task_id, completion_target(), invocation_id="completion_target")
|
|
invalid = CompiledRequirementsSpec.model_validate({"requirements": [{"assumptions": [], "acceptance_claims": [{"claim_kind": "solid_count_equals", "expected": {"value": 0}}]}] * 3})
|
|
before = repository.get_state(task_id)
|
|
result = handler.submit_compiled_spec(task_id, invalid, invocation_id="invalid_spec")
|
|
self.assertIsInstance(result, Rejected)
|
|
self.assertEqual(result.error.code, ErrorCode.REQUIREMENTS_SPEC_INVALID)
|
|
self.assertEqual(repository.get_state(task_id), before)
|
|
|
|
def test_feature_plan_rejection_has_an_independent_retry_budget(self) -> None:
|
|
"""A failed replan must not consume a prior requirements-format retry."""
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
runtime = ProfileCadRuntime(settings(root))
|
|
requirements = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
actions = ActionCommandHandler(repository, artifacts, runtime, RegistryVerifierExecutor(default_registry()))
|
|
workflow = WorkflowCoordinator(
|
|
WorkflowConfig(max_turns=8, format_error_limit=2),
|
|
repository,
|
|
artifacts,
|
|
runtime,
|
|
object(), # The direct retry-budget test does not call a model.
|
|
object(),
|
|
requirements,
|
|
actions,
|
|
)
|
|
initial = repository.create_task("cad_123456abcdef", "Create a plate.")
|
|
documented = transition(initial, "requirements_document_written", requirements_document_path="requirements.md")
|
|
targeted = transition(documented, "completion_target_written", completion_target_path="completion-target.md")
|
|
compiling = transition(targeted, "requirements_compiled", requirements_contract_path="requirements-contract.json")
|
|
scheduled = transition(
|
|
compiling,
|
|
"feature_plan_written",
|
|
feature_plan_path="plans/feature-plan-active.json",
|
|
feature_plan_hash="a" * 64,
|
|
)
|
|
replanning = transition(scheduled, "feature_replan", error=ErrorCode.CANDIDATE_REVIEW_REJECTED)
|
|
self.assertTrue(repository.compare_and_swap(documented))
|
|
self.assertTrue(repository.compare_and_swap(targeted))
|
|
self.assertTrue(repository.compare_and_swap(compiling))
|
|
self.assertTrue(repository.compare_and_swap(scheduled))
|
|
self.assertTrue(repository.compare_and_swap(replanning))
|
|
counters = {"requirements_spec": 1}
|
|
feedback: list[dict[str, object]] = []
|
|
terminal = workflow._requirements_rejection(
|
|
replanning.task_id,
|
|
replanning,
|
|
WorkflowError(ErrorCode.REQUIREMENTS_SPEC_INVALID, "Feature plan violates immutable-node rules."),
|
|
counters,
|
|
feedback,
|
|
tool="write_feature_plan",
|
|
)
|
|
self.assertIsNone(terminal)
|
|
self.assertEqual(repository.get_state(replanning.task_id).phase, TaskPhase.REPLANNING_FEATURE_SUBGRAPH)
|
|
self.assertEqual(counters, {"requirements_spec": 1, "write_feature_plan": 1})
|
|
self.assertEqual(len(feedback), 1)
|
|
|
|
def test_completion_result_reports_frozen_checklist(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
handler = RequirementsCommandHandler(repository, artifacts, default_registry())
|
|
task_id = "cad_123456abcdef"
|
|
repository.create_task(task_id, "Create a coherent flange.")
|
|
artifacts.initialize_task(task_id, "Create a coherent flange.")
|
|
handler.submit_requirements_document(task_id, requirements_document(), invocation_id="requirements_document")
|
|
handler.submit_completion_target(task_id, completion_target(), invocation_id="completion_target")
|
|
handler.submit_compiled_spec(task_id, compiled_flange(), invocation_id="requirements_compile")
|
|
handler.submit_modeling_plan(task_id, modeling_plan(), invocation_id="modeling_plan")
|
|
state = repository.get_state(task_id)
|
|
path = handler.write_completion_result(
|
|
task_id, state,
|
|
claim_results=[{"claim_id": f"claim_{position:03d}", "status": "pass", "evidence": {"measured": True}} for position in range(1, 6)],
|
|
review={"visual_claims": []},
|
|
)
|
|
self.assertEqual(path, "completion-result.md")
|
|
result = (artifacts.task_dir(task_id) / path).read_text(encoding="utf-8")
|
|
target = (artifacts.task_dir(task_id) / "completion-target.md").read_text(encoding="utf-8")
|
|
requirements = (artifacts.task_dir(task_id) / "requirements.md").read_text(encoding="utf-8")
|
|
self.assertIn("Engineering Defaults", requirements)
|
|
self.assertIn("Centered 30 mm through bore", target)
|
|
self.assertIn("Centered 30 mm through bore.: pass", result)
|
|
|
|
def test_sqlite_schema_contains_only_current_requirement_paths(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
database = Path(temporary) / "state.sqlite3"
|
|
SqliteTaskRepository(database)
|
|
with sqlite3.connect(database) as connection:
|
|
columns = {row[1] for row in connection.execute("PRAGMA table_info(tasks)")}
|
|
self.assertIn("requirements_spec_path", columns)
|
|
self.assertIn("requirements_document_path", columns)
|
|
self.assertIn("completion_target_path", columns)
|
|
self.assertIn("modeling_plan_path", columns)
|
|
self.assertIn("clarification_path", columns)
|
|
self.assertNotIn("requirements_draft_path", columns)
|
|
self.assertNotIn("requirements_review_path", columns)
|
|
|
|
def test_role_specific_capability_tools_have_no_review_loop(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
runtime = ProfileCadRuntime(settings(Path(temporary)))
|
|
author = conformance_tools(runtime, role="author")
|
|
reviewer = conformance_tools(runtime, role="reviewer")
|
|
author_names = {item["function"]["name"] for item in author}
|
|
reviewer_names = {item["function"]["name"] for item in reviewer}
|
|
self.assertTrue({"write_requirements_document", "write_completion_target", "compile_requirements_spec", "write_modeling_plan"}.issubset(author_names))
|
|
self.assertNotIn("review_requirements", author_names | reviewer_names)
|
|
self.assertNotIn("get_cdsl_operation_contract", author_names)
|
|
self.assertEqual(reviewer_names, {"observe_images", "review_candidate", "review_final"})
|
|
|
|
def test_capability_cache_is_role_scoped(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
runtime = ProfileCadRuntime(settings(root))
|
|
class Models:
|
|
def __init__(self) -> None:
|
|
self.calls = 0
|
|
async def conformance(self, **_kwargs: object) -> dict[str, object]:
|
|
self.calls += 1
|
|
return {"supported": True, "failures": [], "probe_unavailable": False}
|
|
models = Models()
|
|
first = asyncio.run(verify_model_capability(repository, runtime, models, provider_id="p", model_id="m", role="author"))
|
|
second = asyncio.run(verify_model_capability(repository, runtime, models, provider_id="p", model_id="m", role="author"))
|
|
reviewer = asyncio.run(verify_model_capability(repository, runtime, models, provider_id="p", model_id="m", role="reviewer"))
|
|
self.assertFalse(first.get("cached", False))
|
|
self.assertTrue(second["cached"])
|
|
self.assertNotEqual(first["schema_hash"], reviewer["schema_hash"])
|
|
self.assertEqual(models.calls, 2)
|
|
self.assertIsNotNone(cached_model_capability(repository, runtime, provider_id="p", model_id="m", role="author"))
|
|
|
|
def test_unsupported_capability_is_cached_but_transport_failure_is_not(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
repository = SqliteTaskRepository(root / "state.sqlite3")
|
|
runtime = ProfileCadRuntime(settings(root))
|
|
schema_hash = conformance_hash(conformance_tools(runtime, role="reviewer"), role="reviewer")
|
|
class Unsupported:
|
|
async def conformance(self, **_kwargs: object) -> dict[str, object]:
|
|
return {"supported": False, "failures": [{"message": "schema"}], "probe_unavailable": False}
|
|
asyncio.run(verify_model_capability(repository, runtime, Unsupported(), provider_id="p", model_id="m", role="reviewer"))
|
|
self.assertIsNotNone(repository.model_capability("p", "m", schema_hash))
|
|
class Unavailable:
|
|
async def conformance(self, **_kwargs: object) -> dict[str, object]:
|
|
return {"supported": False, "failures": [{"message": "network"}], "probe_unavailable": True}
|
|
asyncio.run(verify_model_capability(repository, runtime, Unavailable(), provider_id="p2", model_id="m", role="reviewer"))
|
|
self.assertIsNone(repository.model_capability("p2", "m", schema_hash))
|
|
|
|
def test_task_started_is_emitted_before_capability_work(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
configured = settings(root)
|
|
service = AgentService(configured, WorkspaceStore(configured), CdslLibrary(configured))
|
|
message = ChatMessage.model_validate({"id": "user_1", "role": "user", "parts": [{"type": "text", "text": "Create a plate."}]})
|
|
async def first_chunk() -> bytes:
|
|
stream = service.stream([message], None, None)
|
|
chunk = await anext(stream)
|
|
await stream.aclose()
|
|
return chunk
|
|
with patch("app.services.agent_service.verify_model_capability", AsyncMock()) as capability:
|
|
chunk = asyncio.run(first_chunk()).decode("utf-8")
|
|
self.assertIn("task_started", chunk)
|
|
capability.assert_not_awaited()
|
|
self.assertEqual(len(service.v3.repository.running_task_ids()), 1)
|
|
|
|
def test_cached_capabilities_skip_normal_request_probe(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
configured = settings(root)
|
|
service = AgentService(configured, WorkspaceStore(configured), CdslLibrary(configured))
|
|
task_id = "cad_123456abcdef"
|
|
service.v3.workflow.create_task(task_id, "Create a plate.")
|
|
queue: asyncio.Queue = asyncio.Queue()
|
|
with patch("app.services.agent_service.cached_model_capability", return_value={"supported": True}), patch(
|
|
"app.services.agent_service.verify_model_capability", AsyncMock()
|
|
) as verify:
|
|
result = asyncio.run(service._ensure_task_capabilities(
|
|
task_id, ModelIdentity("author", "author-model"), ModelIdentity("reviewer", "reviewer-model"), queue,
|
|
))
|
|
self.assertIsNone(result)
|
|
verify.assert_not_awaited()
|
|
self.assertTrue(queue.empty())
|
|
|
|
def test_missing_cache_is_visible_and_auto_resumes_same_task(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
configured = settings(root)
|
|
service = AgentService(configured, WorkspaceStore(configured), CdslLibrary(configured))
|
|
task_id = "cad_123456abcdef"
|
|
service.v3.workflow.create_task(task_id, "Create a plate.")
|
|
queue: asyncio.Queue = asyncio.Queue()
|
|
with patch("app.services.agent_service.cached_model_capability", return_value=None), patch(
|
|
"app.services.agent_service.verify_model_capability", AsyncMock(return_value={"supported": True, "probe_unavailable": False})
|
|
) as verify:
|
|
result = asyncio.run(service._ensure_task_capabilities(
|
|
task_id, ModelIdentity("author", "author-model"), ModelIdentity("reviewer", "reviewer-model"), queue,
|
|
))
|
|
self.assertIsNone(result)
|
|
self.assertEqual(verify.await_count, 2)
|
|
self.assertEqual(service.v3.repository.get_state(task_id).phase, TaskPhase.DRAFTING_REQUIREMENTS_DOCUMENT)
|
|
events = [queue.get_nowait(), queue.get_nowait()]
|
|
self.assertEqual([item[1]["status"] for item in events], ["waiting", "success"])
|
|
|
|
def test_image_bytes_are_frozen_and_sent_to_vision(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
artifacts = FileArtifactStore(root / "tasks")
|
|
task_id = "cad_123456abcdef"
|
|
source = root / "reference.png"
|
|
png = b"\x89PNG\r\n\x1a\nreference-bytes"
|
|
source.write_bytes(png)
|
|
artifacts.initialize_task(task_id, "Match the image.", image_inputs=[{
|
|
"path": str(source), "mime": "image/png", "sha256": sha256(png).hexdigest(),
|
|
}])
|
|
frozen = Path(artifacts.source_image_paths(task_id)[0])
|
|
class Models:
|
|
def __init__(self) -> None:
|
|
self.messages: list[list[dict[str, object]]] = []
|
|
async def call_tool(self, *, messages: list[dict[str, object]], **_kwargs: object) -> dict[str, object]:
|
|
self.messages.append(messages)
|
|
return {"tool_calls": [], "usage": {}}
|
|
models = Models()
|
|
gateway = RenderedReviewGateway(models)
|
|
tool = {"type": "function", "function": {"name": "observe_images", "parameters": {"type": "object"}}}
|
|
asyncio.run(gateway.review(kind="image_observation", payload={"reference_image_paths": [str(frozen)]}, tool=tool, provider_id="p", model_id="m"))
|
|
image_part = models.messages[0][1]["content"][1]
|
|
encoded = image_part["image_url"]["url"].split(",", 1)[1]
|
|
self.assertEqual(base64.b64decode(encoded), png)
|
|
|
|
def test_final_review_schema_has_ordered_visual_decision_count(self) -> None:
|
|
visual = stateless_final_review_schema(2)["properties"]["visual_claims"]
|
|
self.assertEqual((visual["minItems"], visual["maxItems"]), (2, 2))
|
|
|
|
def test_sqlite_cas_and_outbox_are_atomic(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
repository = SqliteTaskRepository(Path(temporary) / "state.sqlite3")
|
|
initial = repository.create_task("cad_123456abcdef", "Create a plate.")
|
|
changed = transition(initial, "image_observed")
|
|
self.assertTrue(repository.compare_and_swap(changed, events=[{"event": "image_observation_ready"}]))
|
|
self.assertFalse(repository.compare_and_swap(changed, events=[{"event": "duplicate"}]))
|
|
self.assertEqual(len(repository.pending_outbox()), 1)
|
|
delivered = asyncio.run(OutboxDispatcher(repository, IdempotentInProcessPublisher()).dispatch_pending())
|
|
self.assertEqual(len(delivered), 1)
|
|
self.assertEqual(repository.pending_outbox(), [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|