likang/cdsl-cad
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加cadfs_to_cdsl代码，优化cad生成

Browse Source
This commit is contained in:
李康
2026-09-03 10:40:51 +08:00
parent f3eb5dff54
commit c25b73c71e
34 changed files with 3586 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_cad_agent_v3.py
+360 -8
View File
@@ -18,12 +18,13 @@ 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
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,
@@ -42,9 +43,9 @@ 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
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 TaskPhase, TaskState, legal_transitions, retry_resume_event, transition
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
@@ -107,7 +108,10 @@ def completion_target() -> MarkdownDocument:
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}}]},
{"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}}]},
]})
@@ -196,6 +200,194 @@ class CadV3ProtocolTests(unittest.TestCase):
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)))
@@ -445,7 +637,10 @@ class CadV3ProtocolTests(unittest.TestCase):
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}}]},
{"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)
@@ -467,18 +662,125 @@ class CadV3ProtocolTests(unittest.TestCase):
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}}]},
{"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", "visual"])
self.assertEqual([claim["verification_mode"] for claim in claims], ["visual", "deterministic", "visual"])
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)
@@ -572,6 +874,56 @@ class CadV3ProtocolTests(unittest.TestCase):
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)
backend/tests/test_feature_plan.py
+357
View File
@@ -0,0 +1,357 @@
from __future__ import annotations
from pathlib import Path
import sys
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "backend"))
from app.cad_agent.domain.feature_plan import FeaturePlan, FeatureScheduler, node_hash, plan_hash, validate_feature_plan
from app.cad_agent.adapters.artifact_store import FileArtifactStore
from app.cad_agent.adapters.runtime import ProfileCadRuntime
from app.cad_agent.adapters.sqlite_repository import SqliteTaskRepository
from app.cad_agent.adapters.verifier import RegistryVerifierExecutor
from app.cad_agent.application.action_handlers import ActionCommandHandler
from app.cad_agent.application.llm_contracts import CompiledRequirementsSpec, MarkdownDocument
from app.cad_agent.application.requirements import RequirementsCommandHandler
from app.cad_agent.application.results import Accepted, Rejected
from app.cad_agent.domain.state import TaskPhase, transition
from app.cad_agent.domain.verifier_registry import default_registry
from app.settings import ProviderConfig, ProviderModel, Settings
def runtime_settings(root: Path) -> Settings:
provider = ProviderConfig("test", "Test", "https://test.invalid/v1", "key", (ProviderModel("test-model"),))
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=provider.base_url, llm_api_key=provider.api_key, llm_model="test-model", llm_timeout_s=1,
default_provider_id="test", providers=(provider,),
)
def contract() -> dict[str, object]:
return {
"requirements": [
{
"requirement_id": "req_001",
"acceptance_claims": [
{"claim_id": "claim_base", "verification_mode": "deterministic"},
{"claim_id": "claim_visual", "verification_mode": "visual"},
],
},
{
"requirement_id": "req_002",
"acceptance_claims": [{"claim_id": "claim_bore", "verification_mode": "deterministic"}],
},
{
"requirement_id": "req_003",
"acceptance_claims": [{"claim_id": "claim_pattern", "verification_mode": "deterministic"}],
},
],
}
def initial_plan() -> FeaturePlan:
return FeaturePlan.model_validate({
"schema_version": "cad.v3.2.feature-plan.v1",
"parent_plan_hash": "",
"replaces_node_ids": [],
"nodes": [
{"node_id": "base", "priority": 10, "intent": "Create the base.", "atomic_id": "extrude_add_blind", "depends_on": [], "claim_ids": ["claim_base"], "expected_change": "One base solid."},
{"node_id": "bore", "priority": 20, "intent": "Cut the bore.", "atomic_id": "hole_blind", "depends_on": ["base"], "claim_ids": ["claim_bore"], "expected_change": "One through bore."},
{"node_id": "pattern", "priority": 30, "intent": "Add the pattern.", "atomic_id": "hole_blind", "depends_on": ["base"], "claim_ids": ["claim_pattern"], "expected_change": "Mounting holes."},
],
"final_claim_ids": ["claim_visual"],
})
class FeaturePlanTests(unittest.TestCase):
def test_plan_requires_exact_claim_ownership(self) -> None:
plan = initial_plan()
self.assertEqual(validate_feature_plan(plan, contract(), {"extrude_add_blind", "hole_blind"}), [])
broken = plan.model_copy(deep=True)
broken.nodes[1].claim_ids = ["claim_base"]
messages = [item["message"] for item in validate_feature_plan(broken, contract(), {"extrude_add_blind", "hole_blind"})]
self.assertTrue(any("already owned" in message for message in messages))
self.assertTrue(any("has no owner" in message for message in messages))
def test_visual_claim_repeated_on_a_node_is_removed_before_persisting_the_plan(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(), atomic_ids=lambda: ("extrude_add_blind", "hole_blind"))
task_id = "cad_123456abcdef"
repository.create_task(task_id, "Create a part.")
artifacts.initialize_task(task_id, "Create a part.")
self.assertIsInstance(handler.submit_requirements_document(task_id, MarkdownDocument(markdown="Create a part."), invocation_id="requirements"), Accepted)
self.assertIsInstance(handler.submit_completion_target(task_id, MarkdownDocument(markdown="- [ ] A base and a visual edge treatment."), invocation_id="target"), Accepted)
compiled = CompiledRequirementsSpec.model_validate({"requirements": [{"assumptions": [], "acceptance_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "visual", "expected": {"description": "A visible edge treatment."}},
]}]})
self.assertIsInstance(handler.submit_compiled_spec(task_id, compiled, invocation_id="compile"), Accepted)
submitted = FeaturePlan.model_validate({
"schema_version": "cad.v3.2.feature-plan.v1", "parent_plan_hash": "", "replaces_node_ids": [],
"nodes": [{"node_id": "base", "priority": 10, "intent": "Create the base.", "atomic_id": "extrude_add_blind", "depends_on": [], "claim_ids": ["claim_001", "claim_002"], "expected_change": "One base solid."}],
"final_claim_ids": ["claim_002"],
})
accepted = handler.submit_feature_plan(task_id, submitted, invocation_id="plan")
self.assertIsInstance(accepted, Accepted)
state = repository.get_state(task_id)
persisted = artifacts.read_json(task_id, state.feature_plan_path)
self.assertEqual(persisted["nodes"][0]["claim_ids"], ["claim_001"])
self.assertEqual(persisted["final_claim_ids"], ["claim_002"])
def test_unowned_global_health_claim_is_bound_to_the_unique_root_add_feature(self) -> None:
plan = FeaturePlan.model_validate({
"schema_version": "cad.v3.2.feature-plan.v1", "parent_plan_hash": "", "replaces_node_ids": [],
"nodes": [
{"node_id": "base", "priority": 10, "intent": "Create the base.", "atomic_id": "extrude_add_blind", "depends_on": [], "claim_ids": ["claim_bore"], "expected_change": "One base solid."},
{"node_id": "finish", "priority": 20, "intent": "Finish the part.", "atomic_id": "hole_blind", "depends_on": ["base"], "claim_ids": ["claim_pattern"], "expected_change": "One bore."},
],
"final_claim_ids": ["claim_visual"],
})
health_contract = contract()
health_contract["requirements"][0]["acceptance_claims"][0]["claim_kind"] = "single_connected_body"
normalized = RequirementsCommandHandler._assign_unowned_global_health_claims(plan, health_contract)
self.assertEqual(normalized.nodes[0].claim_ids, ["claim_bore", "claim_base"])
self.assertEqual(validate_feature_plan(normalized, health_contract, {"extrude_add_blind", "hole_blind"}), [])
def test_scheduler_uses_ready_nodes_and_fixed_priority(self) -> None:
plan = initial_plan()
scheduler = FeatureScheduler(plan, [])
self.assertEqual(scheduler.next_ready().node_id, "base")
base = plan.nodes[0]
events = [{
"event": "feature_node_verified", "node_id": "base", "node_hash": node_hash(base),
"feature_id": "feature_001", "revision_id": "rev_001",
}]
scheduler = FeatureScheduler(plan, events)
self.assertEqual(scheduler.statuses()["base"], "done")
# Both bore and pattern are ready; priority decides deterministically.
self.assertEqual(scheduler.next_ready().node_id, "bore")
self.assertEqual(scheduler.feature_ids(), {"base": "feature_001"})
def test_revision_cannot_change_done_node_and_replaces_failed_subgraph(self) -> None:
previous = initial_plan()
base = previous.nodes[0]
bore = previous.nodes[1]
events = [
{"event": "feature_node_verified", "node_id": "base", "node_hash": node_hash(base), "feature_id": "feature_001", "revision_id": "rev_001"},
{"event": "feature_node_failed", "node_id": "bore", "node_hash": node_hash(bore), "failure_class": "engine_build", "attempt": 2, "terminal": True},
]
completed = FeatureScheduler(previous, events).completed_node_hashes()
revision = FeaturePlan.model_validate({
"schema_version": "cad.v3.2.feature-plan.v1",
"parent_plan_hash": plan_hash(previous),
"replaces_node_ids": ["bore"],
"nodes": [
base.model_dump(mode="json"),
{"node_id": "bore_revised", "priority": 20, "intent": "Cut the bore with revised operation.", "atomic_id": "hole_blind", "depends_on": ["base"], "claim_ids": ["claim_bore"], "expected_change": "One through bore."},
previous.nodes[2].model_dump(mode="json"),
],
"final_claim_ids": ["claim_visual"],
})
self.assertEqual(
validate_feature_plan(revision, contract(), {"extrude_add_blind", "hole_blind"}, previous_plan=previous, completed_node_hashes=completed, required_replacements={"bore"}),
[],
)
changed = revision.model_copy(deep=True)
changed.nodes[0].intent = "Changed completed node."
self.assertTrue(any("completed node 'base' was modified" in item["message"] for item in validate_feature_plan(changed, contract(), {"extrude_add_blind", "hole_blind"}, previous_plan=previous, completed_node_hashes=completed, required_replacements={"bore"})))
def test_materialized_feature_uses_direct_dag_dependencies_not_previous_history_item(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
runtime = ProfileCadRuntime(runtime_settings(Path(temporary)))
base = {
"schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", "part_id": "dag_test",
"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": "reference_plane", "params": {"plane": {"origin_mm": [0, 0, 5], "normal": [0, 0, 1], "x_dir": [1, 0, 0]}}, "depends_on": ["feature_001"]},
],
}
contract = runtime.operation_contract("reference_axis")
document, audit = runtime.materialize_fragment(
base,
{"feature": {"atomic_id": "reference_axis", "params": {"axis": {"origin_mm": [0, 0, 0], "direction": [0, 0, 1]}}}},
contract,
{},
runtime.reference_tokens(base),
depends_on_feature_ids=("feature_001",),
)
self.assertEqual(document["features"][-1]["depends_on"], ["feature_001"])
self.assertEqual(audit["depends_on_feature_ids"], ["feature_001"])
def test_required_through_cut_gets_a_server_recorded_exit_allowance(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
runtime = ProfileCadRuntime(runtime_settings(Path(temporary)))
base = {
"schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", "part_id": "through_cut",
"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"}],
}
document, audit = runtime.materialize_fragment(
base,
{
"sketch": {
"workplane": {"origin_mm": [0, 0, 5], "normal": [0, 0, 1], "x_dir": [1, 0, 0]},
"profile": {"type": "circle", "center": [0, 0], "radius_mm": 2},
},
"feature": {"atomic_id": "extrude_cut_blind", "params": {"distance_mm": 5, "reverse": True}},
},
runtime.operation_contract("extrude_cut_blind"),
{"body": {"kind": "body", "geometry": {"bbox_mm": [-10, -10, 0, 10, 10, 5]}}},
runtime.reference_tokens(base),
require_through=True,
depends_on_feature_ids=("feature_001",),
)
self.assertAlmostEqual(document["features"][-1]["params"]["distance_mm"], 5.01)
self.assertEqual(audit["server_normalizations"][0]["submitted_mm"], 5.0)
self.assertEqual(audit["server_normalizations"][0]["reason"], "required through-cut exit allowance")
def test_verified_node_publishes_without_a_render_bundle_or_candidate_review(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
registry = default_registry()
runtime = ProfileCadRuntime(runtime_settings(root))
repository = SqliteTaskRepository(root / "state.sqlite3")
artifacts = FileArtifactStore(root / "tasks")
requirements = RequirementsCommandHandler(repository, artifacts, registry, atomic_ids=runtime.supported_atomic_ids)
actions = ActionCommandHandler(repository, artifacts, runtime, RegistryVerifierExecutor(registry))
task_id = "cad_123456abcdef"
repository.create_task(task_id, "Create a disk.")
artifacts.initialize_task(task_id, "Create a disk.")
self.assertIsInstance(requirements.submit_requirements_document(task_id, MarkdownDocument(markdown="Create a disk."), invocation_id="requirements"), Accepted)
self.assertIsInstance(requirements.submit_completion_target(task_id, MarkdownDocument(markdown="- [ ] One connected disk with 20 mm diameter and 5 mm thickness."), invocation_id="target"), Accepted)
compiled = CompiledRequirementsSpec.model_validate({"requirements": [{"assumptions": [], "acceptance_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 20, "axial_span_mm": 5, "tolerance_mm": 0.1}},
]}]})
self.assertIsInstance(requirements.submit_compiled_spec(task_id, compiled, invocation_id="compile"), Accepted)
plan = FeaturePlan.model_validate({
"schema_version": "cad.v3.2.feature-plan.v1", "parent_plan_hash": "", "replaces_node_ids": [],
"nodes": [{"node_id": "base", "priority": 10, "intent": "Create disk.", "atomic_id": "extrude_add_blind", "depends_on": [], "claim_ids": ["claim_001", "claim_002"], "expected_change": "One disk solid."}],
"final_claim_ids": [],
})
self.assertIsInstance(requirements.submit_feature_plan(task_id, plan, invocation_id="plan"), Accepted)
self.assertEqual(repository.get_state(task_id).phase, TaskPhase.SCHEDULING_FEATURE)
self.assertIsInstance(actions.schedule_next_feature(task_id), Accepted)
self.assertEqual(repository.get_state(task_id).phase, TaskPhase.FEATURE_PENDING)
result = actions.submit_feature_fragment(task_id, {
"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}},
}, invocation_id="fragment")
self.assertIsInstance(result, Accepted)
state = repository.get_state(task_id)
self.assertEqual(state.phase, TaskPhase.SCHEDULING_FEATURE)
self.assertEqual(state.active_revision, "rev_001")
self.assertFalse((artifacts.task_dir(task_id) / "revisions" / "rev_001" / "renders").exists())
events = repository.ledger_events(task_id)
self.assertTrue(any(event.get("event") == "feature_node_verified" for event in events))
self.assertFalse(any(event.get("event") == "candidate_built" for event in events))
final_gate = actions.schedule_next_feature(task_id)
self.assertIsInstance(final_gate, Accepted)
self.assertEqual(repository.get_state(task_id).phase, TaskPhase.FINAL_VALIDATION)
self.assertTrue(any(event.get("event") == "feature_plan_complete" for event in repository.ledger_events(task_id)))
def test_feature_plan_schema_binds_plan_revision_lineage_to_state(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
runtime = ProfileCadRuntime(runtime_settings(root))
repository = SqliteTaskRepository(root / "state.sqlite3")
artifacts = FileArtifactStore(root / "tasks")
requirements = RequirementsCommandHandler(repository, artifacts, default_registry(), atomic_ids=runtime.supported_atomic_ids)
task_id = "cad_123456abcdef"
repository.create_task(task_id, "Create a disk.")
artifacts.initialize_task(task_id, "Create a disk.")
self.assertIsInstance(requirements.submit_requirements_document(task_id, MarkdownDocument(markdown="Create a disk."), invocation_id="requirements"), Accepted)
self.assertIsInstance(requirements.submit_completion_target(task_id, MarkdownDocument(markdown="- [ ] One disk."), invocation_id="target"), Accepted)
compiled = CompiledRequirementsSpec.model_validate({"requirements": [{"assumptions": [], "acceptance_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
]}]})
self.assertIsInstance(requirements.submit_compiled_spec(task_id, compiled, invocation_id="compile"), Accepted)
plan = FeaturePlan.model_validate({
"schema_version": "cad.v3.2.feature-plan.v1", "parent_plan_hash": "", "replaces_node_ids": [],
"nodes": [{"node_id": "base", "priority": 10, "intent": "Create disk.", "atomic_id": "extrude_add_blind", "depends_on": [], "claim_ids": ["claim_001"], "expected_change": "One disk solid."}],
"final_claim_ids": [],
})
self.assertIsInstance(requirements.submit_feature_plan(task_id, plan, invocation_id="plan"), Accepted)
scheduled = transition(repository.get_state(task_id), "feature_scheduled")
self.assertTrue(repository.compare_and_swap(scheduled))
replanning = transition(repository.get_state(task_id), "feature_replan")
self.assertTrue(repository.compare_and_swap(replanning))
schema = requirements.feature_plan_schema(task_id)
properties = schema["properties"]
self.assertEqual(properties["parent_plan_hash"]["enum"], [plan_hash(plan)])
self.assertEqual(properties["replaces_node_ids"]["minItems"], 0)
self.assertEqual(properties["replaces_node_ids"]["maxItems"], 0)
node_schema = schema["$defs"]["FeatureNode"]["properties"]
self.assertEqual(node_schema["claim_ids"]["items"], {"enum": ["claim_001"]})
self.assertEqual(properties["final_claim_ids"]["items"], {"enum": []})
def test_pending_assigned_claim_rejects_the_node_checkpoint(self) -> None:
class PendingAssignedClaimVerifier:
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": "pending" if claim.get("claim_id") == "claim_002" else "pass",
"evidence": {"reason": "target geometry has not been introduced"},
}
for claim in claims
]
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
runtime = ProfileCadRuntime(runtime_settings(root))
repository = SqliteTaskRepository(root / "state.sqlite3")
artifacts = FileArtifactStore(root / "tasks")
requirements = RequirementsCommandHandler(repository, artifacts, default_registry(), atomic_ids=runtime.supported_atomic_ids)
actions = ActionCommandHandler(repository, artifacts, runtime, PendingAssignedClaimVerifier())
task_id = "cad_123456abcdef"
repository.create_task(task_id, "Create a disk.")
artifacts.initialize_task(task_id, "Create a disk.")
self.assertIsInstance(requirements.submit_requirements_document(task_id, MarkdownDocument(markdown="Create a disk."), invocation_id="requirements"), Accepted)
self.assertIsInstance(requirements.submit_completion_target(task_id, MarkdownDocument(markdown="- [ ] One disk."), invocation_id="target"), Accepted)
compiled = CompiledRequirementsSpec.model_validate({"requirements": [{"assumptions": [], "acceptance_claims": [
{"claim_kind": "single_connected_body", "expected": {}},
{"claim_kind": "outer_cylindrical_surface", "expected": {"diameter_mm": 20, "tolerance_mm": 0.1}},
]}]})
self.assertIsInstance(requirements.submit_compiled_spec(task_id, compiled, invocation_id="compile"), Accepted)
plan = FeaturePlan.model_validate({
"schema_version": "cad.v3.2.feature-plan.v1", "parent_plan_hash": "", "replaces_node_ids": [],
"nodes": [{"node_id": "base", "priority": 10, "intent": "Create disk.", "atomic_id": "extrude_add_blind", "depends_on": [], "claim_ids": ["claim_001", "claim_002"], "expected_change": "One disk solid."}],
"final_claim_ids": [],
})
self.assertIsInstance(requirements.submit_feature_plan(task_id, plan, invocation_id="plan"), Accepted)
self.assertIsInstance(actions.schedule_next_feature(task_id), Accepted)
result = actions.submit_feature_fragment(task_id, {
"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}},
}, invocation_id="fragment")
self.assertIsInstance(result, Rejected)
self.assertEqual(repository.get_state(task_id).phase, TaskPhase.FEATURE_PENDING)
events = repository.ledger_events(task_id)
failure = next(event for event in reversed(events) if event.get("event") == "feature_node_failed")
self.assertEqual(failure["failure_class"], "node_validation")
self.assertTrue(any(item.get("claim_id") == "claim_002" and item.get("status") == "pending" for item in failure["blockers"]))
self.assertFalse(any(event.get("event") == "feature_node_verified" for event in events))
if __name__ == "__main__":
unittest.main()