"""Execution session state and the geometry adapter boundary. ``ExecutionSession`` owns the active body, body-member graph, replay definitions, and selector-resolution evidence. ``GeometryAdapter`` is the kernel-facing protocol the session consumes; geometry values stay opaque so a different B-rep backend can replace build123d without touching the runtime. """ from __future__ import annotations from copy import deepcopy from dataclasses import dataclass, field from typing import Any, Protocol from .build123d_adapter import Build123dGeometryAdapter from .runtime_base import FeatureExecutionError from .specs import AxisSpec, BendSpec, GearSpec, HoleSpec, PlaneSpec, RackSpec, ThreadSpec, Vector3 from .topology import ( FeaturePlanNode, FeatureResult, RuntimeDiagnostic, SelectorResolution, TopologyDelta, TopologyRecord, TopologyRegistry, validate_selector_provenance_intent, ) class GeometryAdapter(Protocol): """Kernel boundary consumed by the session runtime. Geometry values remain opaque here. A future adapter may use a different B-rep kernel as long as it preserves these construction/query contracts. """ def topology_records(self, body: Any, feature_id: str, body_id: str) -> list[TopologyRecord]: ... def body_solids(self, body: Any) -> list[Any]: ... def body_geometry(self, body: Any) -> dict[str, Any]: ... def surface_geometry(self, surface: Any) -> dict[str, Any]: ... def faces_for_sketch(self, sketch: dict[str, Any]) -> list[Any]: ... def faces_for_sketch_with_source_anchors(self, sketch: dict[str, Any]) -> tuple[list[Any], list[dict[str, Any]]]: ... def face_with_holes(self, outer: Any, holes: list[Any]) -> Any: ... def loft(self, sketches: list[dict[str, Any]]) -> Any: ... def loft_with_topology_delta(self, sketches: list[dict[str, Any]]) -> tuple[Any, TopologyDelta | None]: ... def loft_with_cap_face(self, cap_face: Any, sketches: list[dict[str, Any]]) -> Any: ... def sweep(self, section: Any, spine: Any, *, inner_wires: list[Any] | None = None, make_solid: bool = True, is_frenet: bool = False, transition: Any = None) -> Any: ... def sweep_with_topology_delta(self, section: Any, spine: Any, *, inner_wires: list[Any] | None = None, make_solid: bool = True, is_frenet: bool = False, transition: Any = None) -> tuple[Any, TopologyDelta | None]: ... def sweep_path(self, points: list[Vector3], *, start_tangent: Vector3 | None = None, end_tangent: Vector3 | None = None, parameters: list[float] | None = None) -> Any: ... def face_normal(self, face: Any) -> Vector3: ... def extrude(self, face: Any, direction: Vector3) -> Any: ... def extrude_with_topology_delta(self, face: Any, direction: Vector3) -> tuple[Any, TopologyDelta]: ... def extrude_taper_with_topology_delta(self, face: Any, direction: Vector3, taper_deg: float) -> tuple[Any, TopologyDelta | None]: ... def extrude_taper(self, face: Any, direction: Vector3, taper_deg: float) -> Any: ... def extrude_trimmed(self, face: Any, target: Any, direction: Vector3) -> Any: ... def surface_wires_for_sketch(self, sketch: dict[str, Any]) -> list[Any]: ... def extrude_surface(self, wires: list[Any], direction: Vector3) -> Any: ... def combine_surfaces(self, *surfaces: Any) -> Any: ... def revolve(self, face: Any, angle_deg: float, axis: AxisSpec) -> Any: ... def revolve_surface(self, wire: Any, angle_deg: float, axis: AxisSpec) -> Any: ... def intersect(self, left: Any, right: Any) -> Any: ... def intersect_with_topology_delta(self, left: Any, right: Any) -> tuple[Any, TopologyDelta | None]: ... def transform(self, body: Any, transform: dict[str, Any]) -> Any: ... def transform_with_topology_delta(self, body: Any, transform: dict[str, Any]) -> tuple[Any, TopologyDelta]: ... def fuse(self, body: Any | None, solid: Any) -> Any: ... def fuse_with_topology_delta(self, body: Any | None, solid: Any) -> tuple[Any, TopologyDelta | None]: ... def combine(self, body: Any | None, solid: Any) -> Any: ... def cut(self, body: Any, tool: Any) -> Any: ... def cut_with_topology_delta(self, body: Any, tool: Any) -> tuple[Any, TopologyDelta | None]: ... def sphere(self, radius_mm: float, center_mm: Vector3) -> Any: ... def cylinder_with_topology_delta(self, radius_mm: float, height_mm: float, axis: AxisSpec | None = None) -> tuple[Any, TopologyDelta]: ... def thread_solid(self, spec: ThreadSpec) -> Any: ... def bend_solid(self, spec: BendSpec) -> Any: ... def gear_solid(self, spec: GearSpec) -> Any: ... def rack_solid(self, spec: RackSpec) -> Any: ... def hole_tool(self, spec: HoleSpec, starts: list[Vector3], inward: Vector3, through_depth_mm: float) -> Any: ... def body_center(self, body: Any) -> Vector3: ... def body_span(self, body: Any, direction: Vector3) -> float: ... def vertex_coordinates(self, vertex: Any) -> Vector3: ... def intersection_vertex(self, body: Any, face_sets: list[list[Any]]) -> Any: ... def profile_sample_points(self, face: Any) -> list[Any]: ... def profile_touches_target(self, target: Any, faces: list[Any]) -> bool: ... def next_body_face_after(self, body: Any, faces: list[Any], direction: Vector3, *, excluded_face: Any) -> Any: ... def uniform_intersection_distance(self, target: Any, faces: list[Any], direction: Vector3) -> float: ... def fillet(self, body: Any, radius_mm: float, edges: list[Any]) -> Any: ... def fillet_with_topology_delta(self, body: Any, radius_mm: float, edges: list[Any]) -> tuple[Any, TopologyDelta | None]: ... def tangent_edges(self, body: Any, seeds: list[Any]) -> list[Any]: ... def chamfer(self, body: Any, distance_mm: float, distance_2_mm: float | None, edges: list[Any], face: Any | None = None) -> Any: ... def chamfer_with_topology_delta(self, body: Any, distance_mm: float, distance_2_mm: float | None, edges: list[Any], face: Any | None = None) -> tuple[Any, TopologyDelta | None]: ... def surface_limited_chamfer(self, body: Any, distance_mm: float, edges: list[Any], surfaces: list[Any]) -> Any: ... def shell(self, body: Any, faces: list[Any], thickness_mm: float, *, inward: bool = True) -> Any: ... def shell_with_topology_delta(self, body: Any, faces: list[Any], thickness_mm: float, *, inward: bool = True) -> tuple[Any, TopologyDelta]: ... def export(self, body: Any, path: str) -> None: ... @dataclass class ExecutionSession: sketches: dict[str, dict[str, Any]] nodes: dict[str, FeaturePlanNode] adapter: GeometryAdapter = field(default_factory=Build123dGeometryAdapter) topology: TopologyRegistry = field(default_factory=TopologyRegistry) body: Any | None = None body_id: str | None = None results: dict[str, FeatureResult] = field(default_factory=dict) replay_definitions: dict[str, FeaturePlanNode] = field(default_factory=dict) body_members: dict[str, Any] = field(default_factory=dict) surface_members: dict[str, Any] = field(default_factory=dict) selector_resolutions: list[dict[str, Any]] = field(default_factory=list) active_feature_id: str = "" def register_body( self, feature_id: str, body: Any, *, replay_node: FeaturePlanNode | None = None, body_members: dict[str, Any] | None = None, topology_delta: TopologyDelta | None = None, topology_predecessors: list[TopologyRecord] | None = None, topology_anchors: list[TopologyRecord] | None = None, ) -> None: # #7 multi-body：主体可能是 Compound（多个独立实体，例如两个不相交的 # 拉伸）。body_id 现在反映真实实体结构而不是"最后一个特征的 id"： # 每个独立 Solid 一个 body:{feature}:{index}，供 selector 精确匹配目标 # 实体；单体保持 body:{feature}（与历史行为完全一致）。 self.body = body self.body_id = f"body:{feature_id}" self.body_members = dict(body_members) if body_members is not None else {feature_id: body} # Source-profile anchors are transient construction facts, but unlike # generic role predecessors they must remain addressable by a later # selector intent. They never receive a body id, so active selector # scans cannot mistake them for current model topology. anchors = list(topology_anchors or ()) for anchor in anchors: self.topology.register(anchor) predecessors = [*(topology_predecessors or ()), *anchors] solids = self.adapter.body_solids(body) if len(solids) <= 1: self.topology.replace_body_topology( feature_id, self.body_id, self.adapter.topology_records(body, feature_id, self.body_id), topology_delta=topology_delta, additional_predecessors=predecessors, ) else: # 一个 Compound 的全部成员共享同一个前置 body snapshot。逐个登记会让 # 已登记的本轮成员成为下一个成员的 predecessor，进而把 pattern copy # 的 owner 错误转移到相邻实例。必须原子替换整个多 body 拓扑快照。 members = [ (member_id, self.adapter.topology_records(solid, feature_id, member_id)) for index, solid in enumerate(solids) for member_id in [f"{self.body_id}:{index}"] ] self.topology.replace_body_topologies( feature_id, members, active_body_id=self.body_id, topology_delta=topology_delta, additional_predecessors=predecessors, ) self.topology.register(TopologyRecord( record_id=self.body_id, kind="body", feature_id=feature_id, body_id=self.body_id, geometry=self.adapter.body_geometry(body), value=body, owner_feature_ids=(feature_id,), )) if replay_node is not None: self.replay_definitions[feature_id] = replay_node def register_transient_prism_tool( self, feature_id: str, tool: Any, *, topology_delta: TopologyDelta, topology_anchors: list[TopologyRecord], ) -> list[TopologyRecord]: """Keep one direct-prism primary tool as boolean-input evidence only.""" snapshot_id = f"transient:{feature_id}" records = self.adapter.topology_records(tool, feature_id, snapshot_id) return list(self.topology.register_transient_snapshot( feature_id, snapshot_id, records, topology_delta=topology_delta, anchors=topology_anchors, )) def register_surface(self, feature_id: str, surface: Any) -> str: # 曲面 feature 与实体 body 生命周期相互独立：不能调用 register_body， # 否则 surface 会覆盖 active solid 并改变最终 STEP 的实体结果。 surface_id = f"surface:{feature_id}" self.surface_members[feature_id] = surface for record in self.adapter.topology_records(surface, feature_id, surface_id): self.topology.register(record) self.topology.register(TopologyRecord( record_id=surface_id, kind="surface", feature_id=feature_id, body_id=surface_id, geometry=self.adapter.surface_geometry(surface), value=surface, owner_feature_ids=(feature_id,), )) return surface_id def clear_body(self) -> None: """Clear the active solid after an explicit deleteBodies result.""" self.body = None self.body_id = None self.body_members = {} def _record_selector_resolution(self, resolution: SelectorResolution) -> SelectorResolution: evidence = resolution.as_dict() evidence["feature_id"] = self.active_feature_id self.selector_resolutions.append(evidence) return resolution def _intersection_component_records(self, selector: dict[str, Any]) -> list[TopologyRecord]: matched = selector.get("matched_selectors") if selector.get("match_mode") == "all" else None if matched is not None: if not isinstance(matched, list) or not matched: raise FeatureExecutionError("intersection_selector_unbound", "Intersection selector has no bound face matches") resolved = [self._record_selector_resolution(self.topology.resolve(item, active_body_id=self.body_id)) for item in matched] else: binding_feature_id = selector.get("binding_feature_id") active_body_id = None if binding_feature_id and self.body_id != f"body:{binding_feature_id}" else self.body_id resolved = [self._record_selector_resolution(self.topology.resolve(selector, active_body_id=active_body_id))] failures = [item for item in resolved if item.status != "resolved" or (item.record is None and not item.records)] if failures: detail = failures[0].diagnostic.message if failures[0].diagnostic else "intersection selector component was not resolved" raise FeatureExecutionError("intersection_selector_component_unresolved", detail) return [ record for item in resolved for record in (item.records or ((item.record,) if item.record is not None else ())) ] def _resolve_intersection_vertex(self, selector: dict[str, Any]) -> SelectorResolution: components = selector.get("intersection_of") if self.body is None: return SelectorResolution( selector=selector, status="not_found", candidates=(), diagnostic=RuntimeDiagnostic("missing_extent_body", "Intersection selector requires an existing body"), ) if not isinstance(components, list) or len(components) < 2: return SelectorResolution( selector=selector, status="not_found", candidates=(), diagnostic=RuntimeDiagnostic("intersection_selector_incomplete", "Intersection selector requires at least two face components"), ) try: face_sets = [self._intersection_component_records(component) for component in components] if any(record.kind != "face" for records in face_sets for record in records): raise FeatureExecutionError("intersection_selector_kind", "Intersection selector components must resolve to faces") vertex = self.adapter.intersection_vertex(self.body, [[record.value for record in records] for records in face_sets]) except FeatureExecutionError as error: return SelectorResolution( selector=selector, status="not_found", candidates=(), diagnostic=RuntimeDiagnostic(error.code, str(error), detail=error.detail), ) except ValueError as error: return SelectorResolution( selector=selector, status="not_found", candidates=(), diagnostic=RuntimeDiagnostic("intersection_vertex_unresolved", str(error)), ) point = self.adapter.vertex_coordinates(vertex) record = TopologyRecord( record_id=str(selector.get("stable_id") or f"intersection:{id(vertex)}"), kind="vertex", feature_id=self.active_feature_id, body_id=self.body_id, geometry={"center_mm": list(point)}, value=vertex, owner_feature_ids=tuple(filter(None, [str(selector.get("owner_feature_id") or "")])), ) return SelectorResolution( selector=selector, status="resolved", record=record, candidates=({"score": 1.0, **record.public_dict()},), ) def resolve(self, selector: dict[str, Any]) -> SelectorResolution: if selector.get("intersection_of") is not None: # The session computes vertex intersections directly from resolved # face components, so enforce the same outer provenance gate that # TopologyRegistry.resolve applies before any geometry operation. validation_error = validate_selector_provenance_intent(selector) if validation_error is not None: return self._record_selector_resolution(SelectorResolution( selector=selector, status="not_found", candidates=(), diagnostic=validation_error, )) return self._record_selector_resolution(self._resolve_intersection_vertex(selector)) owner = str(selector.get("owner_feature_id") or "") active_body_id = f"surface:{owner}" if owner in self.surface_members else self.body_id return self._record_selector_resolution(self.topology.resolve(selector, active_body_id=active_body_id)) def result( self, node: FeaturePlanNode, *, context: PlaneSpec | AxisSpec | None = None, diagnostics: list[RuntimeDiagnostic] | None = None, include_body: bool = True, surface_id: str | None = None, ) -> FeatureResult: result = FeatureResult( feature_id=node.feature_id, atomic_id=node.atomic_id, status="executed", body_id=self.body_id if include_body else None, surface_id=surface_id, context=context, replay_definition={"atomic_id": node.atomic_id, "params": deepcopy(node.params), "sketch_id": node.sketch_id}, diagnostics=diagnostics or [], ) self.results[node.feature_id] = result return result def replay_sources(self, source_feature_ids: list[Any]) -> list[FeaturePlanNode]: """Return selected source features in their original history order. A pattern's exported selection order is not an execution order. In particular, a boolean cut may appear before its parent boss in the raw selection array. The CDSL feature list is dependency-ordered by semantic validation, so it is the stable order for replay. """ requested = {str(feature_id) for feature_id in source_feature_ids} sources = [ feature for feature_id, feature in self.nodes.items() if feature_id in requested and feature_id in self.replay_definitions ] if len(sources) != len(requested): missing = sorted(requested - {source.feature_id for source in sources}) raise ValueError(f"pattern source features have no replay definitions: {', '.join(missing)}") return sources