from __future__ import annotations import unittest from engine.cdsl_engine.topology import ( TopologyBlendRelation, TopologyDelta, TopologyDeltaRelation, TopologyLineage, 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"}, }, ], } def _query_expr_leaf(label: str) -> dict: return { "node": "topology_query", "owner": {"node": "literal", "value": "f1.opExtrude"}, "topology_type": {"node": "literal", "value": "SWEPT_FACE"}, "entity_type": {"node": "literal", "value": "FACE"}, "arguments": [{"node": "literal", "value": label}], } def _swept_face_operand(entity_id: str, *, all_fragments: bool = False) -> dict: expression = _query_expr_leaf(entity_id) allowed = ["boundary", "fragment"] if all_fragments else ["boundary"] return { "kind": "face", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("SWEPT_FACE", allowed=allowed, multiplicity="all_fragments" if all_fragments else "one"), "source_entity": {"sketch_id": "F0", "entity_id": entity_id}, "query_expr": {"version": "1.0", "root": expression}, }, } def _proven_set(operands: list[dict], *, operator: str = "union", kind: str = "face") -> dict: contracts = { "union": "proven_operand_union", "intersection": "proven_operand_intersection", "subtraction": "proven_operand_subtraction", } roots = [item["selector_intent"]["query_expr"]["root"] for item in operands] return { "kind": kind, "source": "runtime_snapshot", "confidence": 1.0, "query_operands": operands, "selector_intent": { **_intent("QUERY_SET", allowed=["boundary"], multiplicity="source_qualified"), "query_expr": { "version": "1.0", "root": {"node": "set", "operator": operator, "operands": roots}, }, "query_set_contract": contracts[operator], "set_kind": kind, "body_scope": "active_member", "empty_policy": "reject", "multiple_policy": "all", }, } def _proven_union(operands: list[dict], *, kind: str = "face") -> dict: return _proven_set(operands, kind=kind) def _owner_body_selector(query_input: dict) -> dict: input_root = query_input["selector_intent"]["query_expr"]["root"] return { "kind": "body", "source": "runtime_snapshot", "confidence": 1.0, "query_input": query_input, "selector_intent": { **_intent("OWNER_BODY", allowed=["boundary"]), "kind": "body", "query_expr": { "version": "1.0", "root": { "node": "filter", "filter": "owner_body", "input": input_root, "arguments": [], }, }, "owner_body_contract": "exact_input_owner", "body_scope": "active_member", "empty_policy": "reject", "multiple_policy": "one", }, } def _primary_cut_copy_cap_edge_selector() -> dict: input_root = { "node": "topology_query", "owner": {"node": "literal", "value": "f1.opExtrude"}, "topology_type": {"node": "literal", "value": "CAP_EDGE"}, "entity_type": {"node": "literal", "value": "EDGE"}, "arguments": [], } query_input = { "kind": "edge", "owner_feature_id": "f1", "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.end", "query_expr": {"version": "1.0", "root": input_root}, }, } return { "kind": "edge", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0, "query_input": query_input, "selector_intent": { **_intent("COPY", allowed=["boundary", "continuation"]), "kind": "edge", "copy_contract": "primary_cut_cap_edge", "query_expr": { "version": "1.0", "root": { "node": "topology_query", "owner": {"node": "literal", "value": "f1.boolean.opBoolean"}, "topology_type": {"node": "literal", "value": "COPY"}, "entity_type": {"node": "literal", "value": "EDGE"}, "arguments": [{ "node": "map", "entries": [{"key": "derivedFrom", "value": input_root}], }], }, }, }, } def _primary_cut_copy_cap_face_selector() -> dict: input_root = { "node": "topology_query", "owner": {"node": "literal", "value": "f1.opExtrude"}, "topology_type": {"node": "literal", "value": "CAP_FACE"}, "entity_type": {"node": "literal", "value": "FACE"}, "arguments": [], } query_input = { "kind": "face", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("CAP_FACE", allowed=["boundary", "continuation"]), "kind": "face", "source_entities": [{"sketch_id": "F0", "entity_id": "E0"}], "lineage_role": "extrude.end", "query_expr": {"version": "1.0", "root": input_root}, }, } return { "kind": "face", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0, "query_input": query_input, "selector_intent": { **_intent("COPY", allowed=["boundary", "continuation"]), "kind": "face", "copy_contract": "primary_cut_cap_face_workplane", "query_expr": { "version": "1.0", "root": { "node": "topology_query", "owner": {"node": "literal", "value": "f1.boolean.opBoolean"}, "topology_type": {"node": "literal", "value": "COPY"}, "entity_type": {"node": "literal", "value": "FACE"}, "arguments": [{ "node": "map", "entries": [{"key": "derivedFrom", "value": input_root}], }], }, }, }, } class SelectorProvenanceContractTests(unittest.TestCase): def test_primary_cut_copy_cap_edge_requires_exact_two_link_lineage(self) -> None: registry = TopologyRegistry() source, transient, result = object(), object(), object() registry.register(TopologyRecord( "anchor:f1:edge:0", "edge", "f1", value=source, source_entity=("F0", "E0"), )) registry.register(TopologyRecord( "transient:f1:edge:0", "edge", "f1", "transient:f1", value=transient, transient=True, )) registry.register(TopologyRecord( "body:f1:edge:0", "edge", "f1", "body:f1", value=result, )) registry._lineage.extend([ TopologyLineage( ("anchor:f1:edge:0",), ("transient:f1:edge:0",), "boundary", "kernel_history", "complete", "proven", "extrude", "extrude.end", "f1", "edge", "edge", ), TopologyLineage( ("transient:f1:edge:0",), ("body:f1:edge:0",), "continuation", "kernel_history", "complete", "proven", "subtract", None, "f1", "edge", "edge", ), ]) selector = _primary_cut_copy_cap_edge_selector() resolved = registry.resolve(selector, active_body_id="body:f1") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "copy_lineage") self.assertEqual(resolved.record.record_id, "body:f1:edge:0") self.assertEqual(resolved.evidence["transient_records"], ["transient:f1:edge:0"]) self.assertEqual([item["operation"] for item in resolved.evidence["relations"]], ["extrude", "subtract"]) selector["selector_intent"]["query_expr"]["root"]["arguments"][0]["entries"][0]["value"] = { "node": "literal", "value": "unrelated", } rejected = registry.resolve(selector, active_body_id="body:f1") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_copy_invalid") def test_primary_cut_copy_cap_face_requires_one_complete_active_successor(self) -> None: selector = _primary_cut_copy_cap_face_selector() def registry_for( *, continuation_count: int = 1, continuation_coverage: str = "complete", result_body_id: str = "body:f1", ) -> TopologyRegistry: registry = TopologyRegistry() registry.register(TopologyRecord("transient:f1:face:0", "face", "f1", "transient:f1", transient=True)) registry.register(TopologyRecord("body:f1:face:0", "face", "f1", result_body_id)) registry._lineage.append(TopologyLineage( ("source:f1:profile",), ("transient:f1:face:0",), "boundary", "kernel_history", "complete", "proven", "extrude", "extrude.end", "f1", "face", "face", )) for index in range(continuation_count): registry._lineage.append(TopologyLineage( ("transient:f1:face:0",), ("body:f1:face:0",), "continuation", "kernel_history", continuation_coverage, "proven", "subtract", None, "f1", "face", "face", )) return registry resolved = registry_for().resolve(selector, active_body_id="body:f1") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "copy_lineage") for label, registry in ( ("missing", registry_for(continuation_count=0)), ("ambiguous", registry_for(continuation_count=2)), ("partial", registry_for(continuation_coverage="partial")), ): with self.subTest(continuation=label): rejected = registry.resolve(selector, active_body_id="body:f1") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_copy_kernel_history_missing") inactive = registry_for(result_body_id="body:other").resolve(selector, active_body_id="body:f1") self.assertEqual(inactive.status, "not_found") self.assertEqual(inactive.diagnostic.code, "selector_copy_inactive") def test_primary_cut_copy_swept_face_requires_exact_source_edge_continuation(self) -> None: selector = _primary_cut_copy_cap_face_selector() selector["selector_intent"]["copy_contract"] = "primary_cut_swept_face_workplane" query_input = selector["query_input"] input_intent = query_input["selector_intent"] input_intent["query_family"] = "SWEPT_FACE" input_intent.pop("source_entities") input_intent.pop("lineage_role") input_intent["source_entity"] = {"sketch_id": "F0", "entity_id": "E0"} input_root = input_intent["query_expr"]["root"] input_root["topology_type"] = {"node": "literal", "value": "SWEPT_FACE"} def registry_for(*, continuation_count: int = 1, result_body_id: str = "body:f1") -> TopologyRegistry: registry = TopologyRegistry() registry.register(TopologyRecord( "anchor:f1:edge:0", "edge", "f1", value=object(), source_entity=("F0", "E0"), transient=True, )) registry.register(TopologyRecord("transient:f1:face:0", "face", "f1", "transient:f1", transient=True)) registry.register(TopologyRecord("body:f1:face:0", "face", "f1", result_body_id)) registry._lineage.append(TopologyLineage( ("anchor:f1:edge:0",), ("transient:f1:face:0",), "boundary", "kernel_history", "complete", "proven", "extrude", None, "f1", "edge", "face", )) for _ in range(continuation_count): registry._lineage.append(TopologyLineage( ("transient:f1:face:0",), ("body:f1:face:0",), "continuation", "kernel_history", "complete", "proven", "subtract", None, "f1", "face", "face", )) return registry resolved = registry_for().resolve(selector, active_body_id="body:f1") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "copy_lineage") self.assertEqual(resolved.record.record_id, "body:f1:face:0") for label, registry in (("missing", registry_for(continuation_count=0)), ("ambiguous", registry_for(continuation_count=2))): with self.subTest(continuation=label): rejected = registry.resolve(selector, active_body_id="body:f1") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_copy_kernel_history_missing") inactive = registry_for(result_body_id="body:other").resolve(selector, active_body_id="body:f1") self.assertEqual(inactive.status, "not_found") self.assertEqual(inactive.diagnostic.code, "selector_copy_inactive") def _two_side_registry( self, *, missing_right_result: bool = False, split_left_result: bool = False, right_shares_left_result: bool = False, ) -> TopologyRegistry: registry = TopologyRegistry() left_source, right_source = object(), object() left_result, left_fragment, right_result = object(), object(), object() if right_shares_left_result: right_result = left_result missing_result = object() anchors = [ TopologyRecord("anchor:left", "edge", "f1", value=left_source, source_entity=("F0", "left")), TopologyRecord("anchor:right", "edge", "f1", value=right_source, source_entity=("F0", "right")), ] for anchor in anchors: registry.register(anchor) left_values = (left_result, left_fragment) if split_left_result else (left_result,) right_values = (right_result, missing_result) if missing_right_result else (right_result,) relations = [ TopologyDeltaRelation( "generated", "edge", left_source, left_values, source_kind="edge", result_kind="face", derivation="boundary", ), ] if not right_shares_left_result: relations.append(TopologyDeltaRelation( "generated", "edge", right_source, right_values, source_kind="edge", result_kind="face", derivation="boundary", )) registry.replace_body_topology( "f1", "body:f1", [ TopologyRecord("body:f1:left", "face", "f1", "body:f1", value=left_result), *([TopologyRecord("body:f1:left-fragment", "face", "f1", "body:f1", value=left_fragment)] if split_left_result else []), *([] if right_shares_left_result else [ TopologyRecord("body:f1:right", "face", "f1", "body:f1", value=right_result), ]), ], topology_delta=TopologyDelta("extrude", tuple(relations)), additional_predecessors=anchors, ) if right_shares_left_result: # Set algebra can overlap only when separate complete operations # prove the same active record. Keeping this fact in a different # operation component avoids converting it into an artificial # source merge, which the direct child policy must still reject. registry._lineage.append(TopologyLineage( source_record_ids=("anchor:right",), result_record_ids=("body:f1:left",), derivation="boundary", evidence="kernel_history", coverage="complete", status="proven", operation="independent_extrude", feature_id="f1", source_kind="edge", result_kind="face", )) return registry def test_proven_q_union_preserves_operand_order_and_deduplicates_handles(self) -> None: registry = self._two_side_registry() left = _swept_face_operand("left") right = _swept_face_operand("right") resolved = registry.resolve(_proven_union([left, right, left]), active_body_id="body:f1") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "query_set_union") self.assertEqual([record.record_id for record in resolved.records], ["body:f1:left", "body:f1:right"]) self.assertEqual( [item["result_records"] for item in resolved.evidence["operand_resolutions"]], [["body:f1:left"], ["body:f1:right"], ["body:f1:left"]], ) def test_proven_q_union_keeps_the_complete_fragment_set_per_operand(self) -> None: registry = self._two_side_registry(split_left_result=True) split_left = _swept_face_operand("left", all_fragments=True) right = _swept_face_operand("right") resolved = registry.resolve(_proven_union([split_left, right]), active_body_id="body:f1") self.assertEqual(resolved.status, "resolved") self.assertEqual( [record.record_id for record in resolved.records], ["body:f1:left", "body:f1:left-fragment", "body:f1:right"], ) self.assertEqual( resolved.evidence["operand_resolutions"][0]["result_records"], ["body:f1:left", "body:f1:left-fragment"], ) def test_proven_q_union_evaluates_nested_set_tree_without_flattening(self) -> None: registry = self._two_side_registry() left = _swept_face_operand("left") right = _swept_face_operand("right") nested = _proven_union([right, left]) resolved = registry.resolve( _proven_union([left, nested]), active_body_id="body:f1", ) self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "query_set_union") self.assertEqual( [record.record_id for record in resolved.records], ["body:f1:left", "body:f1:right"], ) self.assertEqual( resolved.evidence["operand_resolutions"][1]["resolution_mode"], "query_set_union", ) def test_proven_q_intersection_keeps_first_operand_order_for_exact_shared_records(self) -> None: registry = self._two_side_registry(split_left_result=True, right_shares_left_result=True) left = _swept_face_operand("left", all_fragments=True) right = _swept_face_operand("right") resolved = registry.resolve( _proven_set([left, right], operator="intersection"), active_body_id="body:f1", ) self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "query_set_intersection") self.assertEqual([record.record_id for record in resolved.records], ["body:f1:left"]) self.assertEqual(resolved.evidence["set_operator"], "intersection") def test_proven_q_subtraction_keeps_left_order_and_removes_exact_right_records(self) -> None: registry = self._two_side_registry(split_left_result=True, right_shares_left_result=True) left = _swept_face_operand("left", all_fragments=True) right = _swept_face_operand("right") resolved = registry.resolve( _proven_set([left, right], operator="subtraction"), active_body_id="body:f1", ) self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "query_set_subtraction") self.assertEqual([record.record_id for record in resolved.records], ["body:f1:left-fragment"]) self.assertEqual(resolved.evidence["set_operator"], "subtraction") def test_proven_q_intersection_and_subtraction_reject_an_empty_final_set(self) -> None: left = _swept_face_operand("left") right = _swept_face_operand("right") cases = ( (_proven_set([left, right], operator="intersection"), self._two_side_registry()), (_proven_set([left, left], operator="subtraction"), self._two_side_registry()), ) for selector, registry in cases: with self.subTest(operator=selector["selector_intent"]["query_expr"]["root"]["operator"]): resolved = registry.resolve(selector, active_body_id="body:f1") self.assertEqual(resolved.status, "not_found") self.assertEqual(resolved.diagnostic.code, "selector_query_empty") def test_proven_q_subtraction_requires_exactly_two_source_operands(self) -> None: left = _swept_face_operand("left") selector = _proven_set([left, left, left], operator="subtraction") resolved = TopologyRegistry().resolve(selector, active_body_id="body:f1") self.assertEqual(resolved.status, "not_found") self.assertEqual(resolved.diagnostic.code, "selector_query_set_invalid") self.assertIn("qSubtraction", resolved.diagnostic.detail["reason"]) def test_proven_q_union_rejects_empty_or_kind_incompatible_contracts(self) -> None: left = _swept_face_operand("left") empty = _proven_union([left, left]) empty["query_operands"] = [] mixed = _proven_union([left, _swept_face_operand("right")]) mixed["query_operands"][1]["kind"] = "edge" non_provenance = _proven_union([left, _swept_face_operand("right")]) non_provenance["query_operands"][1]["geometry"] = {"surface_type": "plane"} for selector, expected in ( (empty, "operands do not match"), (mixed, "parent set kind"), (non_provenance, "unsupported non-provenance evidence"), ): with self.subTest(expected=expected): resolved = TopologyRegistry().resolve(selector, active_body_id="body:f1") self.assertEqual(resolved.status, "not_found") self.assertEqual(resolved.diagnostic.code, "selector_query_set_invalid") self.assertIn(expected, resolved.diagnostic.detail["reason"]) def test_proven_q_union_rejects_inactive_or_partial_operand_lineage(self) -> None: left = _swept_face_operand("left") right = _swept_face_operand("right") selector = _proven_union([left, right]) inactive = self._two_side_registry().resolve(selector, active_body_id="body:later") self.assertEqual(inactive.status, "not_found") self.assertEqual(inactive.diagnostic.code, "selector_body_member_inactive") self.assertEqual(inactive.diagnostic.detail["operand_index"], 0) partial = self._two_side_registry(missing_right_result=True).resolve(selector, active_body_id="body:f1") self.assertEqual(partial.status, "not_found") self.assertEqual(partial.diagnostic.code, "selector_kernel_history_missing") self.assertEqual(partial.diagnostic.detail["operand_index"], 1) 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_q_owner_body_projects_one_proven_active_member_to_its_body_record(self) -> None: registry = self._two_side_registry() body = object() registry.register(TopologyRecord( "body:f1", "body", "f1", "body:f1", value=body, owner_feature_ids=("f1",), )) resolved = registry.resolve( _owner_body_selector(_swept_face_operand("left")), active_body_id="body:f1", ) self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "owner_body") self.assertIs(resolved.record.value, body) self.assertEqual(resolved.evidence["input_record_ids"], ["body:f1:left"]) self.assertEqual(resolved.evidence["body_member"], "body:f1") def test_q_owner_body_rejects_ambiguous_or_missing_exact_body_projection(self) -> None: ambiguous_registry = self._two_side_registry(split_left_result=True) ambiguous_registry.register(TopologyRecord( "body:f1", "body", "f1", "body:f1", value=object(), owner_feature_ids=("f1",), )) ambiguous = ambiguous_registry.resolve( _owner_body_selector(_swept_face_operand("left", all_fragments=True)), active_body_id="body:f1", ) self.assertEqual(ambiguous.status, "not_found") self.assertEqual(ambiguous.diagnostic.code, "selector_owner_body_ambiguous") missing = self._two_side_registry().resolve( _owner_body_selector(_swept_face_operand("left")), active_body_id="body:f1", ) self.assertEqual(missing.status, "not_found") self.assertEqual(missing.diagnostic.code, "selector_owner_body_not_found") def test_q_owner_body_rejects_inactive_input_and_tampered_contract(self) -> None: registry = self._two_side_registry() registry.register(TopologyRecord( "body:f1", "body", "f1", "body:f1", value=object(), owner_feature_ids=("f1",), )) selector = _owner_body_selector(_swept_face_operand("left")) inactive = registry.resolve(selector, active_body_id="body:later") self.assertEqual(inactive.status, "not_found") self.assertEqual(inactive.diagnostic.code, "selector_owner_body_input_unresolved") self.assertEqual(inactive.diagnostic.detail["cause"]["code"], "selector_body_member_inactive") tampered = {**selector, "geometry": {"surface_type": "plane"}} rejected = registry.resolve(tampered, active_body_id="body:f1") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_owner_body_invalid") query_set_child = _owner_body_selector(_swept_face_operand("left")) query_set_child["query_input"] = _proven_union([ _swept_face_operand("left"), _swept_face_operand("left"), ]) query_set_child["selector_intent"]["query_expr"]["root"]["input"] = ( query_set_child["query_input"]["selector_intent"]["query_expr"]["root"] ) rejected_set = registry.resolve(query_set_child, active_body_id="body:f1") self.assertEqual(rejected_set.status, "not_found") self.assertEqual(rejected_set.diagnostic.code, "selector_owner_body_invalid") 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") # A later operation may materialize the same source sketch again. Its # source anchor is not an alternative anchor for this producer query. registry.register(TopologyRecord( "anchor:later", "vertex", "later", source_entities=(("F0", "E0"), ("F0", "E1")), )) replayed_edge_resolution = registry.resolve(edge_selector, active_body_id="body:extrude") self.assertEqual(replayed_edge_resolution.status, "resolved") self.assertEqual(replayed_edge_resolution.record.record_id, edge_resolution.record.record_id) 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) cap_vertex_records = [] for role in ("extrude.start", "extrude.end"): vertex_resolution = registry.resolve({ "kind": "vertex", "owner_feature_id": "extrude", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("CAP_VERTEX", allowed=["boundary"]), "kind": "vertex", "source_entities": [ {"sketch_id": "F0", "entity_id": "E0"}, {"sketch_id": "F0", "entity_id": "E1"}, ], "lineage_role": role, }, }, active_body_id="body:extrude") self.assertEqual(vertex_resolution.status, "resolved") self.assertEqual(vertex_resolution.resolution_mode, "kernel_lineage") self.assertEqual(vertex_resolution.evidence["relations"][0]["source_kind"], "vertex") self.assertEqual(vertex_resolution.evidence["relations"][0]["result_kind"], "vertex") self.assertEqual(vertex_resolution.evidence["relations"][0]["output_role"], role) cap_vertex_records.append(vertex_resolution.record.record_id) self.assertEqual(len(set(cap_vertex_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_exact_preserved_member_keeps_cap_vertex_lineage_across_compound_snapshot(self) -> None: registry = TopologyRegistry() source_vertex = object() cap_vertex = object() other_vertex = object() anchor = TopologyRecord( "anchor:vertex", "vertex", "f1", value=source_vertex, source_entities=(("F0", "E0"), ("F0", "E1")), ) registry.register(anchor) registry.replace_body_topology( "f1", "body:f1", [TopologyRecord("f1:cap", "vertex", "f1", "body:f1", value=cap_vertex)], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation( "generated", "vertex", source_vertex, (cap_vertex,), output_role="extrude.end", source_kind="vertex", result_kind="vertex", ), )), additional_predecessors=[anchor], ) # F3 adds an independent member; F1's cap vertex is exactly the same # B-rep item in its new compound member, while F3 itself is distinct. registry.replace_body_topologies( "f3", [ ("body:f3:0", [TopologyRecord("f3:preserved-cap", "vertex", "f3", "body:f3:0", value=cap_vertex)]), ("body:f3:1", [TopologyRecord("f3:new", "vertex", "f3", "body:f3:1", value=other_vertex)]), ], active_body_id="body:f3", member_preservations=[("body:f1", "body:f3:0")], ) selector = { "kind": "vertex", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("CAP_VERTEX", allowed=["boundary", "continuation"]), "kind": "vertex", "source_entities": [ {"sketch_id": "F0", "entity_id": "E0"}, {"sketch_id": "F0", "entity_id": "E1"}, ], "lineage_role": "extrude.end", }, } resolution = registry.resolve(selector, active_body_id="body:f3") self.assertEqual(resolution.status, "resolved") self.assertEqual(resolution.record.record_id, "f3:preserved-cap") self.assertEqual( [relation["operation"] for relation in resolution.evidence["relations"]], ["extrude", "body_member_preserve"], ) # A merely equal-looking replacement has a new opaque handle and must # not inherit a member-preservation relation. replacement = TopologyRegistry() replacement.register(anchor) replacement.replace_body_topology( "f1", "body:f1", [TopologyRecord("f1:cap", "vertex", "f1", "body:f1", value=cap_vertex)], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation("generated", "vertex", source_vertex, (cap_vertex,), output_role="extrude.end", source_kind="vertex", result_kind="vertex"), )), additional_predecessors=[anchor], ) replacement.replace_body_topologies( "f3", [("body:f3:0", [TopologyRecord("f3:replacement", "vertex", "f3", "body:f3:0", value=object())])], active_body_id="body:f3", member_preservations=[("body:f1", "body:f3:0")], ) rejected = replacement.resolve(selector, active_body_id="body:f3") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_body_member_inactive") 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_circle_profile_uses_only_its_declared_source_anchor(self) -> None: """A direct CDSL circle carries its source identity on the profile.""" from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter sketch = { "id": "profile", "source_sketch_id": "F0", "workplane": { "origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1], }, "profile": { "type": "circle", "center": [0, 0], "radius_mm": 5, "source_entity_id": "E5", }, # This generic entity entry deliberately disagrees. A direct # profile must not look it up or infer the source by geometry. "entities": [{"type": "circle", "center": [0, 0], "radius_mm": 5, "source_entity_id": "wrong"}], } faces, anchors = Build123dGeometryAdapter().faces_for_sketch_with_source_anchors(sketch) self.assertEqual(len(faces), 1) self.assertEqual( {(item["kind"], item.get("source_entity")) for item in anchors}, {("edge", ("F0", "E5"))}, ) unlabelled = {**sketch, "profile": {**sketch["profile"]}} unlabelled["profile"].pop("source_entity_id") _faces, no_anchors = Build123dGeometryAdapter().faces_for_sketch_with_source_anchors(unlabelled) self.assertFalse(no_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_multiregion_imprint_prism_fuse_uses_only_final_history(self) -> None: """A fused IMPRINT region keeps exact outer lineage and rejects a deleted seam.""" from engine.cdsl_engine.build123d_adapter import Build123dGeometryAdapter from engine.cdsl_engine.sketch_solver import resolve_all_sketches profile = { "type": "planar_imprint", "source_entities": [ {"id": "bottom", "curve": {"type": "line", "start": [-5, -5], "end": [5, -5]}}, {"id": "right", "curve": {"type": "line", "start": [5, -5], "end": [5, 5]}}, {"id": "top", "curve": {"type": "line", "start": [5, 5], "end": [-5, 5]}}, {"id": "left", "curve": {"type": "line", "start": [-5, 5], "end": [-5, -5]}}, {"id": "divider", "curve": {"type": "line", "start": [0, -5], "end": [0, 5]}}, ], # A bare source set selects both bounded regions above the divider. "selections": [{"source_entity_id": "top", "face_side": 1}], } resolved = resolve_all_sketches({"geometry": {"sketches": [{ "id": "imprint", "source_sketch_id": "F0", "workplane": {"origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1]}, "profile": profile, }]}}) adapter = Build123dGeometryAdapter() faces, anchor_specs = adapter.faces_for_sketch_with_source_anchors( resolved["geometry"]["sketches"][0], ) self.assertEqual(len(faces), 2) composed = adapter.extrude_faces_with_composed_topology_delta(faces, (0, 0, 2)) self.assertIsNotNone(composed) body, topology_delta = composed self.assertTrue(body.is_valid) self.assertAlmostEqual(float(body.volume), 200.0, places=6) self.assertEqual(topology_delta.history_status, "proven") self.assertEqual(topology_delta.history_reason, "exact_prism_fuse_history") self.assertTrue(all( adapter._is_result_topology_member(body.wrapped, value) for relation in topology_delta.relations for value in relation.result_values )) registry = TopologyRegistry() anchors = [ TopologyRecord( f"anchor:{index}", str(spec["kind"]), "f1", value=spec["value"], source_entity=spec.get("source_entity"), source_entities=tuple(spec.get("source_entities") or ()), ) for index, spec in enumerate(anchor_specs) ] for anchor in anchors: registry.register(anchor) registry.replace_body_topology( "f1", "body:f1", adapter.topology_records(body, "f1", "body:f1"), topology_delta=topology_delta, additional_predecessors=anchors, ) def swept_selector(entity_id: str, multiplicity: str = "all_fragments") -> dict: return { "kind": "face", "owner_feature_id": "f1", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("SWEPT_FACE", allowed=["boundary", "fragment"], multiplicity=multiplicity), "source_entity": {"sketch_id": "F0", "entity_id": entity_id}, }, } outer = registry.resolve(swept_selector("left"), active_body_id="body:f1") self.assertEqual(outer.status, "resolved") self.assertEqual(len(outer.records), 1) split_outer = registry.resolve(swept_selector("top"), active_body_id="body:f1") self.assertEqual(split_outer.status, "resolved") self.assertEqual(len(split_outer.records), 2) # The shared source edge created two intermediate side walls, both # deleted by the exact fuse. It must not acquire a nearby final face. deleted = registry.resolve(swept_selector("divider"), active_body_id="body:f1") self.assertEqual(deleted.status, "not_found") self.assertEqual(deleted.diagnostic.code, "selector_kernel_history_missing") divider_anchor = next(anchor for anchor in anchors if anchor.source_entity == ("F0", "divider")) divider_relation = next( relation for relation in topology_delta.relations if relation.source_kind == "edge" and relation.result_kind == "face" and relation.source_value.IsSame(divider_anchor.value.wrapped) ) self.assertEqual(divider_relation.coverage, "partial") self.assertEqual(divider_relation.status, "unknown") self.assertEqual(divider_relation.result_values, ()) # A split source has two exact physical anchors; the one-result policy # stays ambiguous instead of selecting one fragment by geometry. ambiguous = registry.resolve(swept_selector("top", "one"), active_body_id="body:f1") self.assertEqual(ambiguous.status, "not_found") self.assertEqual(ambiguous.diagnostic.code, "selector_ambiguous") for role in ("extrude.start", "extrude.end"): caps = registry.resolve({ "kind": "face", "owner_feature_id": "f1", "output_role": role, "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("CAP_FACE", allowed=["boundary", "fragment"], multiplicity="all_fragments"), "output_role": role, }, }, active_body_id="body:f1") self.assertEqual(caps.status, "resolved") self.assertEqual(len(caps.records), 2) 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_uses_the_builder_cap_snapshot_not_a_cached_continuation_role(self) -> None: registry = self._source_qualified_section_registry() cap = next(record for record in registry.records() if record.record_id == "f1:cap") continued = object() registry.replace_body_topology("f3", "body:f3", [ TopologyRecord( "f3:continued-cap", "face", "f3", "body:f3", value=continued, owner_feature_ids=("f1",), output_roles=("extrude.end",), ), ], topology_delta=TopologyDelta("subtract", ( TopologyDeltaRelation("preserved", "face", cap.value, (continued,)), ))) 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.resolution_mode, "kernel_intersection") self.assertEqual(resolved.evidence["source_records"][0], "f1:cap") def test_intersect_rejects_duplicate_cap_builder_role_facts(self) -> None: registry = self._source_qualified_section_registry() registry._topology_deltas.append({ "feature_id": "f1", "operation": "extrude", "relations": [{ "output_role": "extrude.end", "coverage": "complete", "lineage_status": "proven", "result_kind": "face", "result_record_ids": ["f1:cap"], }], }) 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_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") def test_blend_transition_evidence_requires_a_complete_three_handle_relation(self) -> None: registry = TopologyRegistry() source_edge, source_face = object(), object() patch_face, modified_face, final_edge = object(), object(), object() registry.replace_body_topology("base", "body:base", [ TopologyRecord("base:edge", "edge", "base", "body:base", value=source_edge), TopologyRecord("base:face", "face", "base", "body:base", value=source_face), ]) registry.replace_body_topology("chamfer", "body:chamfer", [ TopologyRecord("chamfer:patch", "face", "chamfer", "body:chamfer", value=patch_face), TopologyRecord("chamfer:modified", "face", "chamfer", "body:chamfer", value=modified_face), TopologyRecord("chamfer:edge", "edge", "chamfer", "body:chamfer", value=final_edge), ], topology_delta=TopologyDelta("chamfer", blend_relations=( TopologyBlendRelation( source_edge, source_face, patch_face, modified_face, (final_edge,), ), ))) relation = registry.topology_deltas()[0]["relations"][0] self.assertEqual(relation["status"], "exact_blend_boundary") self.assertEqual(relation["source_record_ids"], ["base:edge"]) self.assertEqual(relation["blend_into_source_record_ids"], ["base:face"]) self.assertEqual(relation["blend_into_result_record_ids"], ["chamfer:modified"]) self.assertEqual(relation["patch_face_record_ids"], ["chamfer:patch"]) self.assertEqual(relation["result_record_ids"], ["chamfer:edge"]) self.assertNotIn("base:edge", registry._successors) self.assertFalse(registry.lineage()) incomplete = TopologyRegistry() incomplete.replace_body_topology("base", "body:base", [ TopologyRecord("base:edge", "edge", "base", "body:base", value=source_edge), TopologyRecord("base:face", "face", "base", "body:base", value=source_face), ]) incomplete.replace_body_topology("chamfer", "body:chamfer", [ TopologyRecord("chamfer:patch", "face", "chamfer", "body:chamfer", value=patch_face), TopologyRecord("chamfer:modified", "face", "chamfer", "body:chamfer", value=modified_face), TopologyRecord("chamfer:edge", "edge", "chamfer", "body:chamfer", value=final_edge), ], topology_delta=TopologyDelta("chamfer", blend_relations=( TopologyBlendRelation( source_edge, source_face, patch_face, modified_face, (), coverage="partial", status="unknown", ), ))) self.assertEqual( incomplete.topology_deltas()[0]["relations"][0]["status"], "incomplete_blend_boundary", ) def test_blend_edge_resolver_requires_one_complete_active_source_pair(self) -> None: def registry_with_blends( blend_factory, *, face_role: str = "extrude.end", ) -> TopologyRegistry: registry = TopologyRegistry() anchor_value, cap_edge_value, cap_face_value = object(), object(), object() anchor = TopologyRecord( "anchor:F0:E0", "edge", "f1", value=anchor_value, source_entity=("F0", "E0"), ) registry.register(anchor) registry.replace_body_topology("f1", "body:f1", [ TopologyRecord("f1:cap-edge", "edge", "f1", "body:f1", value=cap_edge_value), TopologyRecord( "f1:cap-face", "face", "f1", "body:f1", value=cap_face_value, output_roles=(face_role,), ), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation( "generated", "edge", anchor_value, (cap_edge_value,), source_kind="edge", result_kind="edge", derivation="boundary", output_role="extrude.end", ), )), additional_predecessors=[anchor]) blends = blend_factory(cap_edge_value, cap_face_value) current = [] for index, blend in enumerate(blends): current.extend(( TopologyRecord(f"f2:patch:{index}", "face", "f2", "body:f2", value=blend.patch_face_value), TopologyRecord(f"f2:modified:{index}", "face", "f2", "body:f2", value=blend.blend_into_result_value), *( TopologyRecord(f"f2:edge:{index}:{edge_index}", "edge", "f2", "body:f2", value=value) for edge_index, value in enumerate(blend.result_values) ), )) registry.replace_body_topology( "f2", "body:f2", current, topology_delta=TopologyDelta("chamfer", blend_relations=blends), ) return registry patch, modified, final_edge = object(), object(), object() # Bind the actual snapshot handles. No geometry and no unrelated # owner can satisfy this resolver path. registry = registry_with_blends( lambda cap_edge, cap_face: ( TopologyBlendRelation(cap_edge, cap_face, patch, modified, (final_edge,)), ), ) selector = { "kind": "edge", "owner_feature_id": "f2", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("BLEND_EDGE", allowed=["boundary"]), "kind": "edge", "evidence": "kernel_history", "blend_sources": { "edge": { "query_family": "CAP_EDGE", "owner_feature_id": "f1", "source_entity": {"sketch_id": "F0", "entity_id": "E0"}, "lineage_role": "extrude.end", }, "face": { "query_family": "CAP_FACE", "owner_feature_id": "f1", "output_role": "extrude.end", }, }, }, } resolved = registry.resolve(selector, active_body_id="body:f2") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "blend_boundary") self.assertEqual(resolved.record.record_id, "f2:edge:0:0") inactive = registry.resolve(selector, active_body_id="body:f1") self.assertEqual(inactive.status, "not_found") self.assertEqual(inactive.diagnostic.code, "selector_relation_non_unique") other_edge = object() ambiguous = registry_with_blends( lambda cap_edge, cap_face: ( TopologyBlendRelation(cap_edge, cap_face, patch, modified, (final_edge,)), TopologyBlendRelation(cap_edge, cap_face, object(), object(), (other_edge,)), ), ).resolve(selector, active_body_id="body:f2") self.assertEqual(ambiguous.status, "not_found") self.assertEqual(ambiguous.diagnostic.code, "selector_relation_non_unique") incomplete = registry_with_blends( lambda cap_edge, cap_face: ( TopologyBlendRelation(cap_edge, cap_face, patch, modified, (), coverage="partial", status="unknown"), ), ).resolve(selector, active_body_id="body:f2") self.assertEqual(incomplete.status, "not_found") self.assertEqual(incomplete.diagnostic.code, "selector_kernel_history_missing") wrong_role = registry_with_blends( lambda cap_edge, cap_face: ( TopologyBlendRelation(cap_edge, cap_face, patch, modified, (final_edge,)), ), face_role="extrude.start", ).resolve(selector, active_body_id="body:f2") self.assertEqual(wrong_role.status, "not_found") self.assertEqual(wrong_role.diagnostic.code, "selector_source_unavailable") def test_blend_edge_resolver_accepts_one_exact_direct_swept_face_source(self) -> None: registry = TopologyRegistry() anchor_value, cap_edge_value, swept_face_value = object(), object(), object() patch, modified, final_edge = object(), object(), object() anchor = TopologyRecord( "anchor:F0:E0", "edge", "f1", value=anchor_value, source_entity=("F0", "E0"), ) registry.register(anchor) registry.replace_body_topology("f1", "body:f1", [ TopologyRecord("f1:cap-edge", "edge", "f1", "body:f1", value=cap_edge_value), TopologyRecord("f1:swept-face", "face", "f1", "body:f1", value=swept_face_value), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation( "generated", "edge", anchor_value, (cap_edge_value,), source_kind="edge", result_kind="edge", derivation="boundary", output_role="extrude.end", ), TopologyDeltaRelation( "generated", "face", anchor_value, (swept_face_value,), source_kind="edge", result_kind="face", derivation="boundary", ), )), additional_predecessors=[anchor]) registry.replace_body_topology("f2", "body:f2", [ TopologyRecord("f2:patch", "face", "f2", "body:f2", value=patch), TopologyRecord("f2:modified", "face", "f2", "body:f2", value=modified), TopologyRecord("f2:edge", "edge", "f2", "body:f2", value=final_edge), ], topology_delta=TopologyDelta("fillet", blend_relations=( TopologyBlendRelation(cap_edge_value, swept_face_value, patch, modified, (final_edge,)), ))) selector = { "kind": "edge", "owner_feature_id": "f2", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("BLEND_EDGE", allowed=["boundary"]), "kind": "edge", "evidence": "kernel_history", "blend_sources": { "edge": { "query_family": "CAP_EDGE", "owner_feature_id": "f1", "source_entity": {"sketch_id": "F0", "entity_id": "E0"}, "lineage_role": "extrude.end", }, "face": { "query_family": "SWEPT_FACE", "owner_feature_id": "f1", "source_entity": {"sketch_id": "F0", "entity_id": "E0"}, }, }, }, } resolved = registry.resolve(selector, active_body_id="body:f2") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "blend_boundary") self.assertEqual(resolved.record.record_id, "f2:edge") selector["selector_intent"]["blend_sources"]["face"]["source_entity"] = { "sketch_id": "F0", "entity_id": "missing", } rejected = registry.resolve(selector, active_body_id="body:f2") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_source_unavailable") def test_blend_face_resolver_requires_one_active_generated_patch(self) -> None: registry = TopologyRegistry() anchor_value, cap_edge_value, patch_value = object(), object(), object() anchor = TopologyRecord( "anchor:F0:E0", "edge", "f1", value=anchor_value, source_entity=("F0", "E0"), ) registry.register(anchor) registry.replace_body_topology("f1", "body:f1", [ TopologyRecord("f1:cap-edge", "edge", "f1", "body:f1", value=cap_edge_value), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation( "generated", "edge", anchor_value, (cap_edge_value,), source_kind="edge", result_kind="edge", derivation="boundary", output_role="extrude.end", ), )), additional_predecessors=[anchor]) registry.replace_body_topology("f2", "body:f2", [ TopologyRecord("f2:patch", "face", "f2", "body:f2", value=patch_value), ], topology_delta=TopologyDelta("chamfer", ( TopologyDeltaRelation( "generated", "edge", cap_edge_value, (patch_value,), source_kind="edge", result_kind="face", derivation="boundary", ), ))) selector = { "kind": "face", "owner_feature_id": "f2", "source": "runtime_snapshot", "confidence": 1.0, "selector_intent": { **_intent("BLEND_FACE", allowed=["boundary"]), "kind": "face", "evidence": "kernel_history", "blend_face_source": { "query_family": "CAP_EDGE", "owner_feature_id": "f1", "source_entity": {"sketch_id": "F0", "entity_id": "E0"}, "lineage_role": "extrude.end", }, }, } resolved = registry.resolve(selector, active_body_id="body:f2") self.assertEqual(resolved.status, "resolved") self.assertEqual(resolved.resolution_mode, "blend_patch_face") self.assertEqual(resolved.record.record_id, "f2:patch") inactive = registry.resolve(selector, active_body_id="body:f1") self.assertEqual(inactive.status, "not_found") self.assertEqual(inactive.diagnostic.code, "selector_relation_non_unique") ambiguous = TopologyRegistry() ambiguous.register(anchor) ambiguous.replace_body_topology("f1", "body:f1", [ TopologyRecord("f1:cap-edge", "edge", "f1", "body:f1", value=cap_edge_value), ], topology_delta=TopologyDelta("extrude", ( TopologyDeltaRelation("generated", "edge", anchor_value, (cap_edge_value,), source_kind="edge", result_kind="edge", derivation="boundary", output_role="extrude.end"), )), additional_predecessors=[anchor]) ambiguous.replace_body_topology("f2", "body:f2", [ TopologyRecord("f2:patch:a", "face", "f2", "body:f2", value=patch_value), TopologyRecord("f2:patch:b", "face", "f2", "body:f2", value=object()), ], topology_delta=TopologyDelta("chamfer", ( TopologyDeltaRelation("generated", "edge", cap_edge_value, (patch_value, object()), source_kind="edge", result_kind="face", derivation="boundary", coverage="partial", status="unknown"), ))) rejected = ambiguous.resolve(selector, active_body_id="body:f2") self.assertEqual(rejected.status, "not_found") self.assertEqual(rejected.diagnostic.code, "selector_kernel_history_missing") if __name__ == "__main__": unittest.main()