from __future__ import annotations import unittest from engine.cdsl_engine.topology import ( TopologyDelta, TopologyDeltaRelation, TopologyRecord, TopologyRegistry, TopologySectionRelation, ) def _intent( family: str = "CAP_FACE", version: str | None = "1511", *, allowed: list[str] | None = None, multiplicity: str = "one", ) -> dict: source_query = { "ast": {"call": family}, "standard_library": "onshape/std/geometry.fs", "standard_library_version": "1511.0", } if version is not None: source_query["featurescript_version"] = version return { "version": "1.0", "kind": "face", "query_family": family, "source_query": source_query, "derivation_policy": {"allowed": allowed or ["continuation"], "multiplicity": multiplicity}, "evidence": "kernel_history", } def _intersection_intent() -> dict: return { "version": "1.0", "kind": "edge", "query_family": "INTERSECT", "source_query": { "ast": {"call": "makeQuery"}, "featurescript_version": "1511", "standard_library": "onshape/std/geometry.fs", "standard_library_version": "1511.0", }, "derivation_policy": {"allowed": ["intersection"], "multiplicity": "one"}, "evidence": "kernel_history", "intersection_sources": [ {"query_family": "CAP_FACE", "owner_feature_id": "f1", "output_role": "extrude.end"}, { "query_family": "SWEPT_FACE", "owner_feature_id": "f2", "source_entity": {"sketch_id": "S2", "entity_id": "E2"}, }, ], } class SelectorProvenanceContractTests(unittest.TestCase): def test_direct_swept_body_resolves_only_the_active_member(self) -> None: """A direct body query is body-graph evidence, never a stable-ID fallback.""" registry = TopologyRegistry() body = object() registry.register(TopologyRecord( "body:f1", "body", "f1", "body:f1", value=body, owner_feature_ids=("f1",), )) selector = { "kind": "body", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("SWEPT_BODY", allowed=["boundary"]), "kind": "body", "evidence": "active_body_member", "body_member_contract": "direct_new_body", }, } resolved = registry.resolve(selector, active_body_id="body:f1") self.assertEqual(resolved.status, "resolved") self.assertIs(resolved.record.value, body) self.assertEqual(resolved.resolution_mode, "body_member") self.assertEqual(resolved.evidence["body_member"], "body:f1") inactive = registry.resolve(selector, active_body_id="body:later") self.assertEqual(inactive.status, "not_found") self.assertEqual(inactive.diagnostic.code, "selector_body_member_inactive") mixed = {**selector, "stable_id": "body:f1"} rejected = registry.resolve(mixed, active_body_id="body:f1") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_source_unavailable") def test_direct_prism_swept_face_requires_proven_continuation_to_active_body(self) -> None: """A side-wall extent may cross only complete kernel continuations.""" registry = TopologyRegistry() source_edge = object() initial_side = object() continued_side = object() anchor = TopologyRecord( "anchor:edge", "edge", "f1", value=source_edge, source_entity=("F0", "E0"), ) registry.register(anchor) registry.replace_body_topology( "f1", "body:f1", [ TopologyRecord("body:f1:face:0", "face", "f1", "body:f1", value=initial_side), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation( "generated", "edge", source_edge, (initial_side,), source_kind="edge", result_kind="face", derivation="boundary", coverage="complete", status="proven", ), )), additional_predecessors=[anchor], ) registry.replace_body_topology( "cut", "body:cut", [ TopologyRecord("body:cut:face:0", "face", "cut", "body:cut", value=continued_side), ], topology_delta=TopologyDelta("subtract", ( TopologyDeltaRelation( "modified", "face", initial_side, (continued_side,), derivation="continuation", coverage="complete", status="proven", ), )), ) selector = { "kind": "face", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("SWEPT_FACE", allowed=["boundary", "continuation"]), "source_entity": {"sketch_id": "F0", "entity_id": "E0"}, }, } resolved = registry.resolve(selector, active_body_id="body:cut") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.record.record_id, "body:cut:face:0") self.assertEqual(resolved.resolution_mode, "kernel_lineage") self.assertEqual( [relation["derivation"] for relation in resolved.evidence["relations"]], ["boundary", "continuation"], ) boundary_only = { **selector, "selector_intent": { **selector["selector_intent"], "derivation_policy": {"allowed": ["boundary"], "multiplicity": "one"}, }, } rejected = registry.resolve(boundary_only, active_body_id="body:cut") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_relation_non_unique") def test_direct_prism_source_anchors_resolve_cross_kind_kernel_lineage(self) -> None: """Direct profile entities resolve only through their prism history.""" from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter adapter = Build123dGeometryAdapter() outer = [ {"type": "line", "start_mm": [0, 0, 0], "end_mm": [10, 0, 0], "source_entity_id": "E0"}, {"type": "line", "start_mm": [10, 0, 0], "end_mm": [10, 5, 0], "source_entity_id": "E1"}, {"type": "line", "start_mm": [10, 5, 0], "end_mm": [0, 5, 0], "source_entity_id": "E2"}, {"type": "line", "start_mm": [0, 5, 0], "end_mm": [0, 0, 0], "source_entity_id": "E3"}, ] faces, anchor_specs = adapter.faces_for_sketch_with_source_anchors({ "id": "profile", "source_sketch_id": "F0", "profile": {"type": "analytic_contours", "contours": []}, "contour_regions_mm": [{"outer": outer, "holes": []}], }) self.assertEqual(len(faces), 1) self.assertEqual( {(item["kind"], item.get("source_entity"), item.get("source_entities")) for item in anchor_specs}, { ("edge", ("F0", "E0"), None), ("edge", ("F0", "E1"), None), ("edge", ("F0", "E2"), None), ("edge", ("F0", "E3"), None), ("vertex", None, (("F0", "E0"), ("F0", "E1"))), ("vertex", None, (("F0", "E0"), ("F0", "E3"))), ("vertex", None, (("F0", "E1"), ("F0", "E2"))), ("vertex", None, (("F0", "E2"), ("F0", "E3"))), }, ) solid, delta = adapter.extrude_with_topology_delta(faces[0], [0, 0, 8]) registry = TopologyRegistry() anchors = [ TopologyRecord( f"anchor:{index}", item["kind"], "extrude", value=item["value"], source_entity=item.get("source_entity"), source_entities=tuple(item.get("source_entities") or ()), ) for index, item in enumerate(anchor_specs) ] for anchor in anchors: registry.register(anchor) registry.replace_body_topology( "extrude", "body:extrude", adapter.topology_records(solid, "extrude", "body:extrude"), topology_delta=delta, additional_predecessors=anchors, ) face_selector = { "kind": "face", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("SWEPT_FACE", allowed=["boundary"]), "source_entity": {"sketch_id": "F0", "entity_id": "E0"}, }, } face_resolution = registry.resolve(face_selector, active_body_id="body:extrude") self.assertEqual(face_resolution.status, "resolved") self.assertEqual(face_resolution.resolution_mode, "kernel_lineage") self.assertEqual(face_resolution.evidence["semantic_anchor"], { "type": "source_entity", "sketch_id": "F0", "entity_id": "E0", }) self.assertEqual(face_resolution.evidence["relations"][0]["source_kind"], "edge") self.assertEqual(face_resolution.evidence["relations"][0]["result_kind"], "face") edge_selector = { "kind": "edge", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("SWEPT_EDGE", allowed=["boundary"]), "source_entities": [ {"sketch_id": "F0", "entity_id": "E0"}, {"sketch_id": "F0", "entity_id": "E1"}, ], }, } edge_resolution = registry.resolve(edge_selector, active_body_id="body:extrude") self.assertEqual(edge_resolution.status, "resolved") self.assertEqual(edge_resolution.evidence["relations"][0]["source_kind"], "vertex") self.assertEqual(edge_resolution.evidence["relations"][0]["result_kind"], "edge") cap_records = [] for role in ("extrude.start", "extrude.end"): cap_resolution = registry.resolve({ "kind": "edge", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("CAP_EDGE", allowed=["boundary"]), "kind": "edge", "source_entity": {"sketch_id": "F0", "entity_id": "E0"}, "lineage_role": role, }, }, active_body_id="body:extrude") self.assertEqual(cap_resolution.status, "resolved") self.assertEqual(cap_resolution.resolution_mode, "kernel_lineage") self.assertEqual(cap_resolution.evidence["relations"][0]["output_role"], role) cap_records.append(cap_resolution.record.record_id) self.assertEqual(len(set(cap_records)), 2) missing_role = registry.resolve({ "kind": "edge", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("CAP_EDGE", allowed=["boundary"]), "kind": "edge", "source_entity": {"sketch_id": "F0", "entity_id": "E0"}, "lineage_role": "extrude.end", }, }, active_body_id="body:other") self.assertEqual(missing_role.status, "not_found") self.assertEqual(missing_role.diagnostic.code, "selector_body_member_inactive") def test_source_anchor_rejects_partial_or_missing_prism_history(self) -> None: registry = TopologyRegistry() source = object() result = object() missing_result = object() anchor = TopologyRecord( "anchor:edge", "edge", "extrude", value=source, source_entity=("F0", "E0"), ) registry.register(anchor) registry.replace_body_topology("extrude", "body:extrude", [ TopologyRecord("extrude:side", "face", "extrude", "body:extrude", value=result), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation( "generated", "edge", source, (result, missing_result), source_kind="edge", result_kind="face", derivation="boundary", coverage="complete", status="proven", ), )), additional_predecessors=[anchor]) selector = { "kind": "face", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("SWEPT_FACE", allowed=["boundary"]), "source_entity": {"sketch_id": "F0", "entity_id": "E0"}, }, } resolution = registry.resolve(selector, active_body_id="body:extrude") # The adapter claimed complete history, but one result is absent from # the final snapshot. The registry must downgrade the relation before # a selector can consume it. relation = registry.topology_deltas()[0]["relations"][0] self.assertEqual(relation["coverage"], "partial") self.assertEqual(relation["result_record_ids"], ["extrude:side"]) self.assertEqual(resolution.status, "not_found") self.assertEqual(resolution.diagnostic.code, "selector_kernel_history_missing") def test_cap_edge_continuation_requires_the_complete_active_chain(self) -> None: """A direct cap edge may cross only a proven one-to-one continuation.""" registry = TopologyRegistry() source_edge = object() initial_cap_edge = object() continued_cap_edge = object() anchor = TopologyRecord( "anchor:edge", "edge", "extrude", value=source_edge, source_entity=("F0", "E0"), ) registry.register(anchor) registry.replace_body_topology( "extrude", "body:extrude", [ TopologyRecord("body:extrude:edge:0", "edge", "extrude", "body:extrude", value=initial_cap_edge), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation( "generated", "edge", source_edge, (initial_cap_edge,), source_kind="edge", result_kind="edge", derivation="boundary", coverage="complete", status="proven", output_role="extrude.start", ), )), additional_predecessors=[anchor], ) registry.replace_body_topology( "cut", "body:cut", [ TopologyRecord("body:cut:edge:0", "edge", "cut", "body:cut", value=continued_cap_edge), ], topology_delta=TopologyDelta("subtract", ( TopologyDeltaRelation( "preserved", "edge", initial_cap_edge, (continued_cap_edge,), source_kind="edge", result_kind="edge", derivation="continuation", coverage="complete", status="proven", ), )), ) selector = { "kind": "edge", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("CAP_EDGE", allowed=["boundary", "continuation"]), "kind": "edge", "source_entity": {"sketch_id": "F0", "entity_id": "E0"}, "lineage_role": "extrude.start", }, } resolved = registry.resolve(selector, active_body_id="body:cut") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.record.record_id, "body:cut:edge:0") self.assertEqual( [relation["derivation"] for relation in resolved.evidence["relations"]], ["boundary", "continuation"], ) boundary_only = { **selector, "selector_intent": { **selector["selector_intent"], "derivation_policy": {"allowed": ["boundary"], "multiplicity": "one"}, }, } rejected = registry.resolve(boundary_only, active_body_id="body:cut") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_relation_non_unique") partial_registry = TopologyRegistry() partial_registry.register(anchor) partial_registry.replace_body_topology( "extrude", "body:extrude", [ TopologyRecord("body:extrude:edge:0", "edge", "extrude", "body:extrude", value=initial_cap_edge), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation( "generated", "edge", source_edge, (initial_cap_edge,), source_kind="edge", result_kind="edge", derivation="boundary", coverage="complete", status="proven", output_role="extrude.start", ), )), additional_predecessors=[anchor], ) partial_registry.replace_body_topology( "cut", "body:cut", [ TopologyRecord("body:cut:edge:0", "edge", "cut", "body:cut", value=continued_cap_edge), ], topology_delta=TopologyDelta("subtract", ( TopologyDeltaRelation( "modified", "edge", initial_cap_edge, (continued_cap_edge,), source_kind="edge", result_kind="edge", derivation="continuation", coverage="partial", status="unknown", ), )), ) partial = partial_registry.resolve(selector, active_body_id="body:cut") self.assertEqual(partial.status, "not_found") self.assertEqual(partial.diagnostic.code, "selector_kernel_history_missing") def test_direct_annulus_source_edges_resolve_each_prism_side_face(self) -> None: """Direct circular hole wires retain distinct source edge provenance.""" from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter adapter = Build123dGeometryAdapter() faces, anchor_specs = adapter.faces_for_sketch_with_source_anchors({ "id": "annulus", "source_sketch_id": "F0", "workplane": { "origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "y_dir": [0, 1, 0], "normal": [0, 0, 1], }, "profile": { "type": "analytic_contours", "contours": [ {"role": "outer", "closed": True, "segments": [{ "type": "circle", "center": [0, 0], "radius_mm": 10, "source_entity_id": "outer", }]}, {"role": "inner", "closed": True, "segments": [{ "type": "circle", "center": [0, 0], "radius_mm": 5, "source_entity_id": "inner", }]}, ], }, }) self.assertEqual(len(faces), 1) self.assertEqual( {(item["kind"], item.get("source_entity")) for item in anchor_specs}, {("edge", ("F0", "outer")), ("edge", ("F0", "inner"))}, ) solid, delta = adapter.extrude_with_topology_delta(faces[0], [0, 0, 8]) registry = TopologyRegistry() anchors = [ TopologyRecord( f"anchor:{index}", item["kind"], "extrude", value=item["value"], source_entity=item.get("source_entity"), ) for index, item in enumerate(anchor_specs) ] for anchor in anchors: registry.register(anchor) registry.replace_body_topology( "extrude", "body:extrude", adapter.topology_records(solid, "extrude", "body:extrude"), topology_delta=delta, additional_predecessors=anchors, ) records = [] for entity_id in ("outer", "inner"): resolution = registry.resolve({ "kind": "face", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("SWEPT_FACE", allowed=["boundary"]), "source_entity": {"sketch_id": "F0", "entity_id": entity_id}, }, }, active_body_id="body:extrude") self.assertEqual(resolution.status, "resolved") self.assertEqual(resolution.resolution_mode, "kernel_lineage") records.append(resolution.record.record_id) self.assertEqual(len(set(records)), 2) def test_direct_linear_profile_with_circle_hole_keeps_exact_prism_anchors(self) -> None: """A direct hole face retains outer vertex and inner edge provenance.""" from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter adapter = Build123dGeometryAdapter() outer = [ {"type": "line", "start_mm": [0, 0, 0], "end_mm": [12, 0, 0], "source_entity_id": "bottom"}, {"type": "line", "start_mm": [12, 0, 0], "end_mm": [12, 8, 0], "source_entity_id": "right"}, {"type": "line", "start_mm": [12, 8, 0], "end_mm": [0, 8, 0], "source_entity_id": "top"}, {"type": "line", "start_mm": [0, 8, 0], "end_mm": [0, 0, 0], "source_entity_id": "left"}, ] inner = [{ "type": "circle", "center_mm": [6, 4, 0], "radius_mm": 1.5, "normal": [0, 0, 1], "x_dir_mm": [1, 0, 0], "source_entity_id": "hole", }] faces, anchor_specs = adapter.faces_for_sketch_with_source_anchors({ "id": "profile", "source_sketch_id": "F0", "profile": {"type": "analytic_contours", "contours": []}, "contour_regions_mm": [{"outer": outer, "holes": [inner]}], }) self.assertEqual(len(faces), 1) self.assertEqual( {(item["kind"], item.get("source_entity")) for item in anchor_specs if item["kind"] == "edge"}, { ("edge", ("F0", "bottom")), ("edge", ("F0", "right")), ("edge", ("F0", "top")), ("edge", ("F0", "left")), ("edge", ("F0", "hole")), }, ) self.assertIn( ("vertex", None, (("F0", "right"), ("F0", "top"))), {(item["kind"], item.get("source_entity"), item.get("source_entities")) for item in anchor_specs}, ) solid, delta = adapter.extrude_with_topology_delta(faces[0], [0, 0, 8]) anchors = [ TopologyRecord( f"anchor:{index}", item["kind"], "extrude", value=item["value"], source_entity=item.get("source_entity"), source_entities=tuple(item.get("source_entities") or ()), ) for index, item in enumerate(anchor_specs) ] registry = TopologyRegistry() for anchor in anchors: registry.register(anchor) registry.replace_body_topology( "extrude", "body:extrude", adapter.topology_records(solid, "extrude", "body:extrude"), topology_delta=delta, additional_predecessors=anchors, ) edge_resolution = registry.resolve({ "kind": "edge", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("SWEPT_EDGE", allowed=["boundary"]), "source_entities": [ {"sketch_id": "F0", "entity_id": "right"}, {"sketch_id": "F0", "entity_id": "top"}, ], }, }, active_body_id="body:extrude") self.assertEqual(edge_resolution.status, "resolved") self.assertEqual(edge_resolution.resolution_mode, "kernel_lineage") self.assertEqual(edge_resolution.evidence["relations"][0]["source_kind"], "vertex") self.assertEqual(edge_resolution.evidence["relations"][0]["result_kind"], "edge") face_resolution = registry.resolve({ "kind": "face", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("SWEPT_FACE", allowed=["boundary"]), "source_entity": {"sketch_id": "F0", "entity_id": "hole"}, }, }, active_body_id="body:extrude") self.assertEqual(face_resolution.status, "resolved") self.assertEqual(face_resolution.resolution_mode, "kernel_lineage") cap_records = [] for role in ("extrude.start", "extrude.end"): cap_resolution = registry.resolve({ "kind": "edge", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("CAP_EDGE", allowed=["boundary"]), "kind": "edge", "source_entity": {"sketch_id": "F0", "entity_id": "hole"}, "lineage_role": role, }, }, active_body_id="body:extrude") self.assertEqual(cap_resolution.status, "resolved") self.assertEqual(cap_resolution.resolution_mode, "kernel_lineage") self.assertEqual(cap_resolution.evidence["relations"][0]["source_kind"], "edge") self.assertEqual(cap_resolution.evidence["relations"][0]["result_kind"], "edge") self.assertEqual(cap_resolution.evidence["relations"][0]["output_role"], role) cap_records.append(cap_resolution.record.record_id) self.assertEqual(len(set(cap_records)), 2) def test_partial_logical_circle_region_does_not_create_a_source_anchor(self) -> None: """A split circle needs every solver-proven segment before restoration.""" from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter from engine.cdsl_engine.sketch_solver import resolve_profile sketch = { "id": "profile", "source_sketch_id": "F0", "workplane": { "origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1], }, "profile": {"type": "analytic_contours", "contours": [ {"closed": True, "segments": [ {"type": "line", "start": [0, 0], "end": [12, 0], "source_entity_id": "bottom"}, {"type": "line", "start": [12, 0], "end": [12, 8], "source_entity_id": "right"}, {"type": "line", "start": [12, 8], "end": [0, 8], "source_entity_id": "top"}, {"type": "line", "start": [0, 8], "end": [0, 0], "source_entity_id": "left"}, ]}, {"closed": True, "segments": [{ "type": "circle", "center": [6, 4], "radius_mm": 1.5, "source_entity_id": "hole", }]}, ]}, } solved = resolve_profile(sketch) solved["contour_regions_mm"][0]["holes"][0][-1].pop("logical_circle_source_entity_id") _faces, anchors = Build123dGeometryAdapter().faces_for_sketch_with_source_anchors(solved) self.assertNotIn( ("edge", ("F0", "hole")), {(item["kind"], item.get("source_entity")) for item in anchors}, ) def test_direct_multiregion_prism_keeps_each_source_side_face(self) -> None: """Each independently built source region keeps its own final evidence.""" from engine.cdsl_engine.runtime import prepare_cdsl_execution source_intent = _intent("SWEPT_FACE", allowed=["boundary"]) cdsl = { "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", "part_id": "two-direct-circles", "meta": {"unit": "mm"}, "geometry": {"sketches": [{ "id": "sketch_F0", "source_sketch_id": "F0", "workplane": { "origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1], }, "profile": {"type": "analytic_contours", "contours": [ {"closed": True, "segments": [{ "type": "circle", "center": [-20, 0], "radius_mm": 5, "source_entity_id": "E0", }]}, {"closed": True, "segments": [{ "type": "circle", "center": [20, 0], "radius_mm": 5, "source_entity_id": "E1", }]}, ]}, }]}, "features": [{ "id": "f_F1", "atomic_id": "extrude_add_blind", "depends_on": [], "sketch_id": "sketch_F0", "params": {"distance_mm": 8, "result_mode": "new_body"}, }], } execution = prepare_cdsl_execution(cdsl) self.assertTrue(execution.analysis.runtime_eligible) execution.execute_all() self.assertEqual(len(execution.session.adapter.body_solids(execution.session.body)), 2) resolutions = {} for entity_id in ("E0", "E1"): resolution = execution.session.topology.resolve({ "kind": "face", "owner_feature_id": "f_F1", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **source_intent, "source_entity": {"sketch_id": "F0", "entity_id": entity_id}, }, }, active_body_id=execution.session.body_id) self.assertEqual(resolution.status, "resolved") self.assertEqual(resolution.resolution_mode, "kernel_lineage") self.assertIsNotNone(resolution.record) resolutions[entity_id] = resolution self.assertNotEqual( resolutions["E0"].record.record_id, resolutions["E1"].record.record_id, ) self.assertNotEqual( resolutions["E0"].evidence["body_member"], resolutions["E1"].evidence["body_member"], ) active_records = { record.record_id for record in execution.session.topology.records() if record.body_id and record.body_id.startswith(f"{execution.session.body_id}:") } side_relations = [ relation for delta in execution.session.topology.topology_deltas() for relation in delta["relations"] if relation["source_kind"] == "edge" and relation["result_kind"] == "face" ] self.assertEqual(len(side_relations), 2) self.assertTrue(all( relation["coverage"] == "complete" and relation["lineage_status"] == "proven" and len(relation["result_record_ids"]) == 1 and set(relation["result_record_ids"]).issubset(active_records) for relation in side_relations )) def test_unknown_source_version_is_preserved_then_rejected_by_resolver(self) -> None: registry = TopologyRegistry() selector = { "kind": "face", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intent(version=None), } resolution = registry.resolve(selector) self.assertEqual(resolution.status, "not_found") self.assertEqual(resolution.diagnostic.code, "selector_query_version_unknown") def test_known_but_unregistered_query_version_is_rejected(self) -> None: registry = TopologyRegistry() selector = { "kind": "face", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intent(version="1512"), } resolution = registry.resolve(selector) self.assertEqual(resolution.status, "not_found") self.assertEqual(resolution.diagnostic.code, "selector_query_unsupported") self.assertEqual(resolution.diagnostic.detail["verified_versions"], ["1511"]) def test_missing_or_unregistered_standard_library_revision_is_rejected(self) -> None: registry = TopologyRegistry() selector = { "kind": "face", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intent(), } missing = {**selector, "selector_intent": {**selector["selector_intent"], "source_query": { "ast": {"call": "CAP_FACE"}, "featurescript_version": "1511", }}} missing_resolution = registry.resolve(missing) self.assertEqual(missing_resolution.status, "not_found") self.assertEqual(missing_resolution.diagnostic.code, "selector_query_version_unknown") unregistered = {**selector, "selector_intent": {**selector["selector_intent"], "source_query": { **selector["selector_intent"]["source_query"], "standard_library_version": "1512.0", }}} unregistered_resolution = registry.resolve(unregistered) self.assertEqual(unregistered_resolution.status, "not_found") self.assertEqual(unregistered_resolution.diagnostic.code, "selector_query_unsupported") self.assertEqual( unregistered_resolution.diagnostic.detail["verified_standard_libraries"], [{"path": "onshape/std/geometry.fs", "version": "1511.0"}], ) def test_unregistered_query_intent_cannot_fall_back_to_stable_id_or_geometry(self) -> None: registry = TopologyRegistry() face = object() registry.replace_body_topology("base", "body:base", [ TopologyRecord( "base:face", "face", "base", "body:base", value=face, geometry={"center_mm": [0.0, 0.0, 0.0]}, ), ]) selector = { "kind": "face", "owner_feature_id": "base", "stable_id": "base:face", "source": "runtime_snapshot", "confidence": 1.0, "geometry": {"center_mm": [0.0, 0.0, 0.0]}, "selector_intent": _intent("LOFT_EDGE_CHAIN", version="1793"), } resolution = registry.resolve(selector, active_body_id="body:base") self.assertEqual(resolution.status, "not_found") self.assertEqual(resolution.diagnostic.code, "selector_query_unsupported") self.assertEqual(resolution.diagnostic.detail["query_family"], "LOFT_EDGE_CHAIN") self.assertEqual(resolution.diagnostic.detail["verified_versions"], []) def test_active_stable_id_does_not_replace_provenance_evidence(self) -> None: registry = TopologyRegistry() source = object() registry.replace_body_topology("base", "body:base", [ TopologyRecord("base:face", "face", "base", "body:base", value=source), ]) selector = { "kind": "face", "owner_feature_id": "base", "stable_id": "base:face", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intent(), } resolution = registry.resolve(selector, active_body_id="body:base") self.assertEqual(resolution.status, "not_found") self.assertEqual(resolution.diagnostic.code, "selector_kernel_history_missing") def test_output_role_requires_a_relation_allowed_by_its_policy(self) -> None: registry = TopologyRegistry() profile = object() cap = object() registry.replace_body_topology("profile", "body:profile", [ TopologyRecord("profile:face", "face", "profile", "body:profile", value=profile), ]) registry.replace_body_topology("extrude", "body:extrude", [ TopologyRecord("extrude:cap", "face", "extrude", "body:extrude", value=cap), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation("generated", "face", profile, (cap,), output_role="extrude.end"), ))) selector = { "kind": "face", "owner_feature_id": "extrude", "output_role": "extrude.end", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intent(allowed=["continuation"]), } rejected = registry.resolve(selector, active_body_id="body:extrude") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_kernel_history_missing") selector["selector_intent"] = _intent(allowed=["boundary"]) resolved = registry.resolve(selector, active_body_id="body:extrude") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "operation_role") self.assertEqual(resolved.evidence["result_records"], ["extrude:cap"]) self.assertEqual(resolved.evidence["relations"][0]["derivation"], "boundary") def test_all_fragments_rejects_a_missing_branch(self) -> None: registry = TopologyRegistry() source = object() first = object() missing_second = object() registry.replace_body_topology("base", "body:base", [ TopologyRecord("base:face", "face", "base", "body:base", value=source), ]) registry.replace_body_topology("split", "body:split", [ TopologyRecord("split:first", "face", "split", "body:split", value=first), ], topology_delta=TopologyDelta("split", ( TopologyDeltaRelation("modified", "face", source, (first, missing_second)), ))) selector = { "kind": "face", "owner_feature_id": "base", "stable_id": "base:face", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intent(allowed=["fragment"], multiplicity="all_fragments"), } resolution = registry.resolve(selector, active_body_id="body:split") self.assertEqual(resolution.status, "not_found") self.assertEqual(resolution.diagnostic.code, "selector_kernel_history_missing") def test_merge_requires_merge_policy_even_with_one_final_record(self) -> None: registry = TopologyRegistry() first = object() second = object() merged = object() registry.replace_body_topology("base", "body:base", [ TopologyRecord("base:first", "face", "base", "body:base", value=first), TopologyRecord("base:second", "face", "base", "body:base", value=second), ]) registry.replace_body_topology("boolean", "body:boolean", [ TopologyRecord("boolean:merged", "face", "boolean", "body:boolean", value=merged), ], topology_delta=TopologyDelta("fuse", ( TopologyDeltaRelation("modified", "face", first, (merged,)), TopologyDeltaRelation("modified", "face", second, (merged,)), ))) selector = { "kind": "face", "owner_feature_id": "base", "stable_id": "base:first", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intent(allowed=["continuation"]), } rejected = registry.resolve(selector, active_body_id="body:boolean") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_relation_non_unique") selector["selector_intent"] = _intent(allowed=["continuation", "merge"]) resolved = registry.resolve(selector, active_body_id="body:boolean") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.record.record_id, "boolean:merged") def test_output_role_does_not_bypass_operation_merge_policy(self) -> None: registry = TopologyRegistry() first = object() second = object() cap = object() registry.replace_body_topology("profiles", "body:profiles", [ TopologyRecord("profiles:first", "face", "profiles", "body:profiles", value=first), TopologyRecord("profiles:second", "face", "profiles", "body:profiles", value=second), ]) registry.replace_body_topology("extrude", "body:extrude", [ TopologyRecord("extrude:cap", "face", "extrude", "body:extrude", value=cap), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation("generated", "face", first, (cap,), output_role="extrude.end"), TopologyDeltaRelation("generated", "face", second, (cap,)), ))) selector = { "kind": "face", "owner_feature_id": "extrude", "output_role": "extrude.end", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intent(allowed=["boundary"]), } rejected = registry.resolve(selector, active_body_id="body:extrude") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_relation_non_unique") selector["selector_intent"] = _intent(allowed=["boundary", "merge"]) resolved = registry.resolve(selector, active_body_id="body:extrude") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.record.record_id, "extrude:cap") def test_cross_kind_relation_keeps_distinct_source_and_result_evidence(self) -> None: registry = TopologyRegistry() vertex = object() edge = object() registry.replace_body_topology("base", "body:base", [ TopologyRecord("base:vertex", "vertex", "base", "body:base", value=vertex), ]) registry.replace_body_topology("fillet", "body:fillet", [ TopologyRecord("fillet:edge", "edge", "fillet", "body:fillet", value=edge), ], topology_delta=TopologyDelta("fillet", ( TopologyDeltaRelation( "generated", "edge", vertex, (edge,), source_kind="vertex", result_kind="edge", ), ))) relation = registry.topology_deltas()[0]["relations"][0] lineage = registry.lineage()[0] self.assertEqual(relation["source_record_ids"], ["base:vertex"]) self.assertEqual(relation["result_record_ids"], ["fillet:edge"]) self.assertEqual((relation["source_kind"], relation["result_kind"]), ("vertex", "edge")) self.assertEqual((lineage.source_kind, lineage.result_kind), ("vertex", "edge")) def _source_qualified_section_registry(self, *, sections: int = 1, qualified: bool = True) -> TopologyRegistry: registry = TopologyRegistry() raw_cap, cap = object(), object() anchor, swept = object(), object() registry.replace_body_topology("seed", "body:seed", [ TopologyRecord("seed:cap", "face", "seed", "body:seed", {"center_mm": [-1, 0, 0]}, raw_cap), ]) registry.replace_body_topology("f1", "body:f1", [ TopologyRecord("f1:cap", "face", "f1", "body:f1", {"center_mm": [0, 0, 0]}, cap), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation("generated", "face", raw_cap, (cap,), output_role="extrude.end"), ))) source_anchor = TopologyRecord( "anchor:f2", "edge", "f2", value=anchor, source_entity=("S2", "E2"), ) registry.register(source_anchor) registry.replace_body_topology("f2", "body:f2", [ TopologyRecord("f2:swept", "face", "f2", "body:f2", {"center_mm": [1, 0, 0]}, swept), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation( "generated", "edge", anchor, (swept,), source_kind="edge", result_kind="face", derivation="boundary", ), )), additional_predecessors=[source_anchor]) result_edges = [object() for _index in range(sections)] registry.replace_body_topology("boolean", "body:boolean", [ TopologyRecord(f"boolean:section:{index}", "edge", "boolean", "body:boolean", value=edge) for index, edge in enumerate(result_edges) ], topology_delta=TopologyDelta( "subtract", section_values=tuple(result_edges) if not qualified else (), section_relations=( tuple(TopologySectionRelation((cap, swept), edge) for edge in result_edges) if qualified else () ), ), additional_predecessors=[ TopologyRecord("f1:cap", "face", "f1", "body:f1", {"center_mm": [0, 0, 0]}, cap, output_roles=("extrude.end",)), ]) return registry def test_intersect_resolves_only_one_source_qualified_section_edge(self) -> None: registry = self._source_qualified_section_registry() resolved = registry.resolve({ "kind": "edge", "owner_feature_id": "boolean", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intersection_intent(), }, active_body_id="body:boolean") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.record.record_id, "boolean:section:0") self.assertEqual(resolved.resolution_mode, "kernel_intersection") section_relation = next(item for item in resolved.evidence["relations"] if item["derivation"] == "intersection") self.assertEqual(set(section_relation["source_record_ids"]), {"f1:cap", "f2:swept"}) def test_intersect_rejects_unqualified_section_edge(self) -> None: registry = self._source_qualified_section_registry(qualified=False) result = registry.resolve({ "kind": "edge", "owner_feature_id": "boolean", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intersection_intent(), }, active_body_id="body:boolean") self.assertEqual(result.status, "not_found") self.assertEqual(result.diagnostic.code, "selector_kernel_history_missing") def test_intersect_rejects_multiple_source_qualified_section_edges(self) -> None: registry = self._source_qualified_section_registry(sections=2) result = registry.resolve({ "kind": "edge", "owner_feature_id": "boolean", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intersection_intent(), }, active_body_id="body:boolean") self.assertEqual(result.status, "not_found") self.assertEqual(result.diagnostic.code, "selector_relation_non_unique") def test_session_intersection_vertex_uses_the_common_provenance_gate(self) -> None: """Deferred provenance must not reach the session geometry shortcut.""" from engine.cdsl_engine.session import ExecutionSession class RecordingAdapter: def __init__(self) -> None: self.intersection_calls = 0 def intersection_vertex(self, body, face_sets): self.intersection_calls += 1 return object() def vertex_coordinates(self, vertex): return (1.0, 2.0, 3.0) adapter = RecordingAdapter() session = ExecutionSession(sketches={}, nodes={}, adapter=adapter) session.body = object() session.body_id = "body:base" session._intersection_component_records = lambda _selector: [TopologyRecord( "base:face", "face", "base", "body:base", value=object(), )] base_selector = { "kind": "vertex", "owner_feature_id": "cut", "intersection_of": [{"kind": "face"}, {"kind": "face"}], } for intent, expected_code in ( (_intent("INTERSECT", allowed=["intersection"], multiplicity="none"), "selector_query_unsupported"), (_intent("INTERSECT", version="1793", allowed=["intersection"]), "selector_query_unsupported"), (_intent("INTERSECT", version=None, allowed=["intersection"]), "selector_query_version_unknown"), ): with self.subTest(intent=intent): result = session.resolve({**base_selector, "selector_intent": intent}) self.assertEqual(result.status, "not_found") self.assertEqual(result.diagnostic.code, expected_code) self.assertEqual(adapter.intersection_calls, 0) legacy_result = session.resolve(base_selector) self.assertEqual(legacy_result.status, "resolved") self.assertEqual(adapter.intersection_calls, 1) def test_transient_prism_snapshot_is_historical_not_selectable(self) -> None: registry = TopologyRegistry() source_edge = object() tool_face = object() anchor = TopologyRecord( "anchor:tool:E0", "edge", "tool", value=source_edge, source_entity=("tool-sketch", "E0"), ) registered = registry.register_transient_snapshot( "tool", "transient:tool", [TopologyRecord("transient:tool:face:0", "face", "tool", "transient:tool", value=tool_face)], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation( "generated", "edge", source_edge, (tool_face,), source_kind="edge", result_kind="face", derivation="boundary", ), )), anchors=[anchor], ) self.assertTrue(all(record.transient for record in registered)) self.assertTrue(all(record.transient for record in registry.records())) self.assertEqual(registry.topology_deltas()[0]["output_snapshot_ids"], ["transient:tool"]) self.assertEqual( registry.resolve({"kind": "face", "owner_feature_id": "tool"}).status, "not_found", ) with self.assertRaisesRegex(ValueError, "complete direct-prism history"): registry.register_transient_snapshot( "partial-tool", "transient:partial-tool", [TopologyRecord("transient:partial-tool:face:0", "face", "partial-tool", value=object())], topology_delta=TopologyDelta("extrude", history_status="partial"), anchors=[anchor], ) def test_intersect_rejects_section_edge_outside_active_member(self) -> None: registry = self._source_qualified_section_registry() result = registry.resolve({ "kind": "edge", "owner_feature_id": "boolean", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": _intersection_intent(), }, active_body_id="body:later") self.assertEqual(result.status, "not_found") self.assertEqual(result.diagnostic.code, "selector_body_member_inactive") if __name__ == "__main__": unittest.main()