From 9e5aa48c3d2a176c1941ca7d53f24804f41fe68a Mon Sep 17 00:00:00 2001 From: ganjihong Date: Thu, 27 Aug 2026 18:08:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(engine):=20=E6=94=AF=E6=8C=81=E5=A4=9A?= =?UTF-8?q?=E5=AE=9E=E4=BD=93=20body=20=E4=B8=8E=20selector=20=E6=8C=81?= =?UTF-8?q?=E4=B9=85=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 多实体主体拆分为独立 body_id，支持前缀匹配解析 - 重建结果新增 solid_count 字段 - fillet/chamfer 后 selector stable_id 可解析到演化后继 --- .../engine/cdsl_engine/build123d_adapter.py | 10 + backend/engine/cdsl_engine/runtime.py | 20 +- backend/engine/cdsl_engine/runtime_types.py | 186 +++++++++-- .../tests/test_engine_multi_body_contract.py | 304 ++++++++++++++++++ .../tests/test_engine_selector_persistence.py | 221 +++++++++++++ 5 files changed, 716 insertions(+), 25 deletions(-) create mode 100644 backend/tests/test_engine_multi_body_contract.py create mode 100644 backend/tests/test_engine_selector_persistence.py diff --git a/backend/engine/cdsl_engine/build123d_adapter.py b/backend/engine/cdsl_engine/build123d_adapter.py index fe8558fd..343caaaf 100644 --- a/backend/engine/cdsl_engine/build123d_adapter.py +++ b/backend/engine/cdsl_engine/build123d_adapter.py @@ -385,6 +385,16 @@ class Build123dGeometryAdapter: # 将主体导出为 STEP 文件。 export_step(body, path) + @staticmethod + def body_solids(body: Any) -> list[Any]: + # 提取主体内的全部独立 Solid：Compound 返回成员，单个 Solid 返回自身。 + # build123d 对部分退化布尔结果可能抛异常，退化为把主体整体视为一个实体。 + try: + solids = list(body.solids()) + except Exception: + return [body] if body is not None else [] + return solids or ([body] if body is not None else []) + @staticmethod def body_geometry(body: Any) -> dict[str, Any]: # 汇总主体基本几何信息：包围盒与体积。 diff --git a/backend/engine/cdsl_engine/runtime.py b/backend/engine/cdsl_engine/runtime.py index 1f502218..37097331 100644 --- a/backend/engine/cdsl_engine/runtime.py +++ b/backend/engine/cdsl_engine/runtime.py @@ -74,6 +74,7 @@ class GeometryAdapter(Protocol): """ 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 faces_for_sketch(self, sketch: dict[str, Any]) -> list[Any]: ... def extrude(self, face: Any, direction: Vector3) -> Any: ... @@ -108,9 +109,23 @@ class ExecutionSession: active_feature_id: str = "" def register_body(self, feature_id: str, body: Any, *, replay_node: FeaturePlanNode | 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.topology.replace_body_topology(feature_id, self.body_id, self.adapter.topology_records(body, feature_id, self.body_id)) + 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)) + else: + for index, solid in enumerate(solids): + member_id = f"{self.body_id}:{index}" + self.topology.replace_body_topology( + feature_id, member_id, + self.adapter.topology_records(solid, feature_id, member_id), + active_body_id=self.body_id, + ) 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,), @@ -967,6 +982,9 @@ def rebuild_cdsl(cdsl: dict[str, Any], out_step: Path, *, strict: bool = True) - "out_step": str(out_step), "volume_mm3": float(geometry["volume_mm3"]), "bbox_mm": {"min": bbox[:3], "max": bbox[3:]}, + # #7 multi-body：重建结果里的独立实体数（Compound 成员数）， + # 与 batch 验证的 document_truth.geometry.solid_body_count 对齐。 + "solid_count": len(session.adapter.body_solids(session.body)), "feature_results": [result.as_dict() for result in session.results.values()], "runtime_diagnostics": [diagnostic.as_dict() for diagnostic in diagnostics], "topology_records": [record.public_dict() for record in session.topology.records()], diff --git a/backend/engine/cdsl_engine/runtime_types.py b/backend/engine/cdsl_engine/runtime_types.py index def85b0a..0c2a7a4c 100644 --- a/backend/engine/cdsl_engine/runtime_types.py +++ b/backend/engine/cdsl_engine/runtime_types.py @@ -423,6 +423,11 @@ class TopologyRegistry: self._records: list[TopologyRecord] = [] self._by_feature: dict[str, list[TopologyRecord]] = {} self._active_body_id: str | None = None + # #8 selector 持久性：old_record_id -> [new_record_id]。fillet/chamfer + # 会把一条直线边拆分为若干段（中间直段 + 两端圆弧），旧边不再与任何 + # 新边几何等价；这里记录"位置轨迹延续"的直段后继，使后续 selector 的 + # stable_id 引用可以解析到 active body 内的新形态。 + self._successors: dict[str, list[str]] = {} def register(self, record: TopologyRecord) -> None: self._records.append(record) @@ -446,7 +451,10 @@ class TopologyRegistry: self.register(record) return record - def replace_body_topology(self, feature_id: str, body_id: str, records: Iterable[TopologyRecord]) -> None: + def replace_body_topology( + self, feature_id: str, body_id: str, records: Iterable[TopologyRecord], + *, active_body_id: str | None = None, + ) -> None: """Record a fresh B-rep snapshot after a feature mutates the body. OCC topology object identity is invalidated by most body mutations. @@ -455,11 +463,21 @@ class TopologyRegistry: object has one geometrically equivalent predecessor. A changed or split object intentionally becomes owned by this feature instead of being guessed as belonging to an older one. + + ``active_body_id`` names the whole-body group when ``body_id`` is a + member of a multi-solid body (issue #7): the group id keeps the next + mutation's predecessor lookup scoped to every solid of the previous + body, while each member keeps its own ``body:{feature}:{index}`` id. """ previous = [ record for record in self._records - if self._active_body_id is not None and record.body_id == self._active_body_id + if self._active_body_id is not None and record.body_id is not None + and ( + record.body_id == self._active_body_id + or record.body_id.startswith(f"{self._active_body_id}:") + ) ] + records = list(records) consumed_predecessors: set[str] = set() for record in records: predecessor = self._unique_equivalent_predecessor(record, previous, consumed_predecessors) @@ -477,7 +495,27 @@ class TopologyRegistry: owner_feature_ids=owners, ) ) - self._active_body_id = body_id + # #8 selector 持久性：被消费（拆分成段）的旧边记录演化后继，供后续 + # selector 的 stable_id 引用解析到 active body 内的新形态。多条演化 + # 候选时只登记"漂移显著最小"的那条（例如底面边圆角后既有缩短的直段 + # 也有圆角过渡带的新边，前者的端点与原边重合、漂移更小）；漂移并列 + # （如竖直边被完整消费成两条等距直段）属于本质歧义，保守不登记。 + for prior in previous: + if prior.record_id in consumed_predecessors: + continue + candidates = sorted( + ( + (self._evolved_drift(prior, record), record.record_id) + for record in records if self._evolved_equivalent(prior, record) + ), + key=lambda item: item[0], + ) + if not candidates: + continue + best, second = candidates[0], (candidates[1] if len(candidates) > 1 else None) + if second is None or (second[0] - best[0]) > max(0.5, 0.2 * best[0]): + self._successors[prior.record_id] = [best[1]] + self._active_body_id = active_body_id or body_id @staticmethod def _numbers_equal(left: Any, right: Any, *, tolerance: float = 1e-6) -> bool: @@ -555,6 +593,56 @@ class TopologyRegistry: ] return matches[0] if len(matches) == 1 else None + @staticmethod + def _evolved_drift(prior: TopologyRecord, current: TopologyRecord) -> float | None: + """Endpoint drift between direction-aligned straight edges. + + Returns the minimum total endpoint drift (mm) when the two edges are + collinear straight lines (either orientation), otherwise ``None``. + """ + if prior.kind != current.kind: + return None + left, right = prior.geometry, current.geometry + if left.get("curve_type") != "line" or right.get("curve_type") != "line": + return None + if None in (left.get("start_mm"), left.get("end_mm"), right.get("start_mm"), right.get("end_mm")): + return None + + def _delta(a: tuple[float, float, float], b: tuple[float, float, float]) -> tuple[float, float, float]: + return (b[0] - a[0], b[1] - a[1], b[2] - a[2]) + + def _dist(a: tuple[float, float, float], b: tuple[float, float, float]) -> float: + return sqrt(sum((a[i] - b[i]) ** 2 for i in range(3))) + + left_dir = _delta(left["start_mm"], left["end_mm"]) + right_dir = _delta(right["start_mm"], right["end_mm"]) + if _length(left_dir) <= 1e-9 or _length(right_dir) <= 1e-9: + return None + cross = ( + left_dir[1] * right_dir[2] - left_dir[2] * right_dir[1], + left_dir[2] * right_dir[0] - left_dir[0] * right_dir[2], + left_dir[0] * right_dir[1] - left_dir[1] * right_dir[0], + ) + if _length(cross) / (_length(left_dir) * _length(right_dir)) > 1e-3: + return None + same_order = _dist(left["start_mm"], right["start_mm"]) + _dist(left["end_mm"], right["end_mm"]) + reversed_order = _dist(left["start_mm"], right["end_mm"]) + _dist(left["end_mm"], right["start_mm"]) + return min(same_order, reversed_order) + + @classmethod + def _evolved_equivalent(cls, prior: TopologyRecord, current: TopologyRecord, *, drift_mm: float = 5.0) -> bool: + """Loose "position trajectory" equivalence used for evolved successors. + + Unlike ``_geometry_equivalent`` (strict, anti-false-positive provenance), + this deliberately tolerates small endpoint drift: fillet/chamfer split a + straight edge into segments (a middle straight run plus end arcs). The + straight run keeps the same direction and stays within ``drift_mm`` of the + original edge, so it can serve as the edge's evolved successor. Uniqueness + is enforced by the caller (only a single best candidate is recorded). + """ + drift = cls._evolved_drift(prior, current) + return drift is not None and drift <= drift_mm + @staticmethod def _vector_score(expected: Any, actual: Any, tolerance: float = 1e-4) -> float | None: try: @@ -579,8 +667,14 @@ class TopologyRegistry: for key in ("surface_type", "curve_type"): if key in selector_geometry: if record_geometry.get(key) != selector_geometry[key]: - return None - scores.append(1.0) + # #8 selector 持久性：fillet/chamfer 会把直线边演化为圆弧、 + # 平面演化为柱面，但被选中拓扑的位置锚定（bbox/center/端点） + # 不变。曲线/曲面类型变化不再一票否决，而是记低分：位置完全 + # 重合的候选（同一条边的形态演化）仍可胜出；位置不重合的 + # 相邻边会被 0 分项拉低，仍被 minimum_score 挡住。 + scores.append(0.5) + else: + scores.append(1.0) if "bbox_mm" in selector_geometry: expected = selector_geometry["bbox_mm"] actual = record_geometry.get("bbox_mm") @@ -615,7 +709,14 @@ class TopologyRegistry: owner = selector.get("owner_feature_id") candidates = [record for record in self._records if record.kind == kind] if active_body_id and kind in {"face", "edge", "vertex", "body"}: - candidates = [record for record in candidates if record.body_id == active_body_id] + # #7 multi-body：记录 body_id 可能是 body:{feature}:{index}（多体 + # 成员），用前缀匹配把整个主体的记录纳入候选，同时保证旧 body 的 + # 记录（不同 feature 前缀）不会泄漏进来。 + candidates = [ + record for record in candidates + if record.body_id == active_body_id + or (record.body_id is not None and record.body_id.startswith(f"{active_body_id}:")) + ] if owner: candidates = [record for record in candidates if owner in record.owners] geometry = normalize_selector_geometry(selector.get("geometry")) @@ -632,31 +733,68 @@ class TopologyRegistry: ) stable_id = str(selector.get("stable_id") or "").strip() if stable_id: - exact = [record for record in candidates if record.record_id == stable_id] + # #8 selector 持久性：stable_id 是跨 body 演化的持久标识符，精确 + # 匹配在 active body 过滤之前对整个记录集（kind + owner 过滤）执行。 + # 命中已过期（旧 body）的记录时，经演化后继映射解析到 active body + # 内的新形态（fillet/chamfer 拆段后的直段后继）；无后继则回落到 + # 几何打分流程。 + stable_records = [ + record for record in self._records + if record.kind == kind and (not owner or owner in record.owners) + ] + exact = [record for record in stable_records if record.record_id == stable_id] if len(exact) == 1: record = exact[0] - # A stable ID is only a lookup accelerator for snapshot-aware - # selectors. It cannot revive a B-rep entity whose geometric - # signature changed after an upstream rebuild. - if selector.get("snapshot_id"): - score = self._geometry_score(geometry, record.geometry) if geometry else None - if score is None or score < minimum_score: + is_active = active_body_id is None or ( + record.body_id == active_body_id + or (record.body_id is not None and record.body_id.startswith(f"{active_body_id}:")) + ) + if not is_active: + successors = [ + candidate for candidate in stable_records + if candidate.record_id in self._successors.get(record.record_id, ()) + and ( + candidate.body_id == active_body_id + or (candidate.body_id is not None and active_body_id and candidate.body_id.startswith(f"{active_body_id}:")) + ) + ] + if len(successors) == 1: + record = successors[0] + is_active = True + elif len(successors) > 1: return SelectorResolution( selector=selector, - status="not_found", - candidates=({"score": round(float(score or 0), 6), **record.public_dict()},), + status="ambiguous", + candidates=tuple({"score": 1.0, **candidate.public_dict()} for candidate in successors), diagnostic=RuntimeDiagnostic( - code="selector_geometry_mismatch", - message="The stable selector record no longer matches its geometry signature", - detail={"stable_id": stable_id, "score": score, "minimum_score": minimum_score}, + code="selector_ambiguous", + message="More than one evolved successor record satisfies the stable_id", + detail={"stable_id": stable_id, "candidate_count": len(successors)}, ), ) - return SelectorResolution( - selector=selector, - status="resolved", - record=record, - candidates=({"score": round(float(score), 6) if selector.get("snapshot_id") else 1.0, **record.public_dict()},), - ) + if is_active: + # A stable ID is only a lookup accelerator for snapshot-aware + # selectors. It cannot revive a B-rep entity whose geometric + # signature changed after an upstream rebuild. + if selector.get("snapshot_id"): + score = self._geometry_score(geometry, record.geometry) if geometry else None + if score is None or score < minimum_score: + return SelectorResolution( + selector=selector, + status="not_found", + candidates=({"score": round(float(score or 0), 6), **record.public_dict()},), + diagnostic=RuntimeDiagnostic( + code="selector_geometry_mismatch", + message="The stable selector record no longer matches its geometry signature", + detail={"stable_id": stable_id, "score": score, "minimum_score": minimum_score}, + ), + ) + return SelectorResolution( + selector=selector, + status="resolved", + record=record, + candidates=({"score": round(float(score), 6) if selector.get("snapshot_id") else 1.0, **record.public_dict()},), + ) if len(exact) > 1: return SelectorResolution( selector=selector, diff --git a/backend/tests/test_engine_multi_body_contract.py b/backend/tests/test_engine_multi_body_contract.py new file mode 100644 index 00000000..60bc3206 --- /dev/null +++ b/backend/tests/test_engine_multi_body_contract.py @@ -0,0 +1,304 @@ +"""#7 ExecutionSession 单 active body → 多实体 body_id 回归测试。 + +中文说明 +-------- +这个文件在测试什么（issue #7「ExecutionSession 单 active body」的回归测试）： + + 1. 背景：ExecutionSession 只跟踪一个 active body（session.body / + session.body_id = body:{feature_id}）。多实体零件（例如两个不相交 + 的拉伸 add，布尔并后 build123d 返回 Compound，含 ≥2 个独立 Solid） + 被当成一个 body 处理： + - 拓扑记录把 Compound 的所有面/边/顶点统一登记为一个 body_id； + - body_id 实际上变成"最后执行的特征的 id"，不反映真实实体数； + - 后续特征无法精确匹配"某个实体"上的拓扑，多体信息在重建报告里 + 完全丢失。 + 修复后（#7）：register_body 通过 adapter.body_solids 拆出独立实体， + 每个 Solid 一个 body:{feature}:{index}，selector resolve 与拓扑继承 + 用前缀匹配整个主体；单体路径保持 body:{feature} 完全不变。 + + 2. 本测试套件把"多体可识别、单体不受影响"固定下来： + - 多体契约：两个不相交 add → 最后主体含 ≥2 个独立 body_id， + rebuild 输出 solid_count=2； + - 多体 + 后续操作契约：在多体主体上打孔/切除仍 resolve 到正确实体； + - 单体护栏：单个 add → 只有 1 个实体、solid_count=1，body_id 语义 + 与修复前一致。 + + 3. sys.path 说明：把 backend 与 backend/engine 加入搜索路径，直接 import + cdsl_engine 包做端到端测试（与既有测试风格一致）。 + +函数功能一览 +------------ + _workplane(origin) 构造指定原点的 XY 平面草图工作平面。 + _rectangle(minimum, maximum) 构造 XY 平面内的矩形轮廓（2D 多边形）。 + _single_boss_doc() 10×10×10 单体拉伸文档（体积 1000）。 + _two_boss_doc() 两个不相交 10×10×10 拉伸（x 相距 20）， + 布尔并后为 2 个独立 Solid（体积 2000）。 + MultiBodyContractTests 见各测试方法 docstring。 +""" + +from __future__ import annotations + +import math +import sys +import tempfile +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "backend")) +sys.path.insert(0, str(ROOT / "backend" / "engine")) + +from cdsl_engine.runtime import rebuild_cdsl # noqa: E402 + + +# --------------------------------------------------------------------------- +# 测试夹具 +# --------------------------------------------------------------------------- + + +def _workplane(*, origin: list[float]) -> dict: + return {"origin_mm": origin, "x_dir": [1, 0, 0], "normal": [0, 0, 1]} + + +def _rectangle(minimum: list[float], maximum: list[float]) -> dict: + return {"type": "polygon", "vertices": [ + [minimum[0], minimum[1]], [maximum[0], minimum[1]], + [maximum[0], maximum[1]], [minimum[0], maximum[1]], + ]} + + +def _single_boss_doc() -> dict: + return { + "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", + "part_id": "multi-body-single", "meta": {"unit": "mm"}, + "geometry": {"sketches": [{ + "id": "s1", "workplane": _workplane(origin=[0, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5]), + }]}, + "features": [ + {"id": "add_1", "atomic_id": "extrude_add_blind", "depends_on": [], + "params": {"distance_mm": 10}, "sketch_id": "s1"}, + ], + } + + +def _two_boss_doc() -> dict: + """两个不相交 10×10×10 拉伸：主体 x∈[-5,5]，第二个 x∈[15,25]。""" + return { + "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", + "part_id": "multi-body-two", "meta": {"unit": "mm"}, + "geometry": {"sketches": [ + {"id": "s1", "workplane": _workplane(origin=[0, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5])}, + {"id": "s2", "workplane": _workplane(origin=[20, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5])}, + ]}, + "features": [ + {"id": "add_1", "atomic_id": "extrude_add_blind", "depends_on": [], + "params": {"distance_mm": 10}, "sketch_id": "s1"}, + {"id": "add_2", "atomic_id": "extrude_add_blind", "depends_on": ["add_1"], + "params": {"distance_mm": 10}, "sketch_id": "s2"}, + ], + } + + +# --------------------------------------------------------------------------- +# 测试套件 +# --------------------------------------------------------------------------- + + +class MultiBodyContractTests(unittest.TestCase): + """多体识别「实体数-拓扑归属-单体护栏」三方合同测试。""" + + def test_two_disjoint_bosses_report_two_independent_body_ids(self) -> None: + """多体契约：两个不相交 add → 最后主体含 ≥2 个独立 body_id。 + + 修复前 body_id 是"每特征一 id"（body:add_1 / body:add_2），Compound + 的所有面都登记为 body:add_2，无法区分独立实体。修复后 register_body + 为每个 Solid 分配 body:{feature}:{index}，最后主体应同时包含 + body:add_2:0 与 body:add_2:1 的拓扑记录。 + """ + cdsl = _two_boss_doc() + with tempfile.TemporaryDirectory() as directory: + rebuilt = rebuild_cdsl(cdsl, Path(directory) / "two.step") + body_ids = sorted({r["body_id"] for r in rebuilt["topology_records"] if r.get("body_id")}) + self.assertIn("body:add_2:0", body_ids) + self.assertIn("body:add_2:1", body_ids) + # 两个成员都要有可 resolve 的面记录（而不是只有整体 body 记录）。 + member_faces = { + r["body_id"] for r in rebuilt["topology_records"] + if r["kind"] == "face" and r.get("body_id", "").startswith("body:add_2:") + } + self.assertEqual(member_faces, {"body:add_2:0", "body:add_2:1"}) + self.assertAlmostEqual(rebuilt["volume_mm3"], 2000.0, places=5) + + def test_rebuild_reports_solid_count(self) -> None: + """多体契约：rebuild 输出 solid_count 反映独立实体数。""" + with tempfile.TemporaryDirectory() as directory: + two = rebuild_cdsl(_two_boss_doc(), Path(directory) / "two.step") + one = rebuild_cdsl(_single_boss_doc(), Path(directory) / "one.step") + self.assertEqual(two["solid_count"], 2) + self.assertEqual(one["solid_count"], 1) + + def test_single_body_keeps_legacy_body_id(self) -> None: + """单体护栏：单个 add → 1 个实体，body_id 语义与修复前一致。 + + 防止把多体拆分做成"所有主体都拆"：单体主体必须保持 + body:{feature}（无 :index 后缀），后续 selector 行为不变。 + """ + with tempfile.TemporaryDirectory() as directory: + rebuilt = rebuild_cdsl(_single_boss_doc(), Path(directory) / "one.step") + body_ids = sorted({r["body_id"] for r in rebuilt["topology_records"] if r.get("body_id")}) + self.assertEqual(body_ids, ["body:add_1"]) + face_ids = { + r["body_id"] for r in rebuilt["topology_records"] + if r["kind"] == "face" and r.get("body_id") + } + self.assertEqual(face_ids, {"body:add_1"}) + + def test_cut_on_multi_body_mutates_only_the_intersected_solid(self) -> None: + """多体 + 后续操作契约：多体主体上切除仍 resolve 到正确实体。 + + 两个不相交 box（各 1000）。在第一个 box（x∈[-5,5]）顶面打贯穿孔 + （r=1、深 10）：只有第一个 box 被切，第二个 box（x∈[15,25]）不受 + 影响。期望体积 = 2000 − π×1²×10，且 STEP 仍含 2 个 Solid。 + """ + base = _two_boss_doc() + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + baseline = rebuild_cdsl(base, root / "baseline.step") + top_face = next( + item for item in baseline["topology_records"] + if item["kind"] == "face" + and item["geometry"]["surface_type"] == "plane" + and item["geometry"]["normal"][2] > 0.9 + and abs(item["geometry"]["center_mm"][0]) < 1.0 + ) + with_cut = { + "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", + "part_id": "multi-body-cut", "meta": {"unit": "mm"}, + "geometry": {"sketches": [ + {"id": "s1", "workplane": _workplane(origin=[0, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5])}, + {"id": "s2", "workplane": _workplane(origin=[20, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5])}, + {"id": "cut", "workplane": _workplane(origin=[0, 0, 0]), + "profile": {"type": "circle", "center": [0, 0], "radius_mm": 1}}, + ]}, + "features": [ + {"id": "add_1", "atomic_id": "extrude_add_blind", "depends_on": [], + "params": {"distance_mm": 10}, "sketch_id": "s1"}, + {"id": "add_2", "atomic_id": "extrude_add_blind", "depends_on": ["add_1"], + "params": {"distance_mm": 10}, "sketch_id": "s2"}, + {"id": "cut_1", "atomic_id": "extrude_cut_blind", "depends_on": ["add_2"], + "params": {"distance_mm": 10}, "sketch_id": "cut"}, + ], + } + rebuilt = rebuild_cdsl(with_cut, root / "cut.step") + + self.assertAlmostEqual(rebuilt["volume_mm3"], 2000 - math.pi * 10, places=5) + self.assertEqual(rebuilt["solid_count"], 2) + # 最后主体（cut 后）仍是 2 个独立实体，每个都有拓扑记录。 + member_ids = { + r["body_id"] for r in rebuilt["topology_records"] + if r["kind"] == "face" and r.get("body_id", "").startswith("body:cut_1:") + } + self.assertEqual(member_ids, {"body:cut_1:0", "body:cut_1:1"}) + + def test_face_selector_resolves_on_multi_body_via_prefix_matching(self) -> None: + """多体 + selector resolve 契约：face selector 经前缀匹配命中正确实体。 + + 这是 resolve 前缀匹配的哨兵测试：两个不相交 box 合并为 Compound 后， + 拓扑记录属于 body:add_2:0 / body:add_2:1。若 resolve 仍用精确 body_id + 过滤（active_body_id=body:add_2 不匹配任何记录），host face 会解析失败； + 只有前缀匹配才能让 hole 的宿主面命中第一个 box 的顶面。体积 = 2000 − 10π + 且 box2（x∈[15,25]）不受影响，证明解析到的确实是正确实体。 + """ + base = _two_boss_doc() + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + baseline = rebuild_cdsl(base, root / "baseline.step") + # 第一个 box 的顶面：center ≈ (0, 0, 10)。 + top_face = next( + item for item in baseline["topology_records"] + if item["kind"] == "face" + and item["geometry"]["surface_type"] == "plane" + and item["geometry"]["normal"][2] > 0.9 + and abs(item["geometry"]["center_mm"][0]) < 1.0 + ) + with_hole = { + "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", + "part_id": "multi-body-hole", "meta": {"unit": "mm"}, + "geometry": {"sketches": [ + {"id": "s1", "workplane": _workplane(origin=[0, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5])}, + {"id": "s2", "workplane": _workplane(origin=[20, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5])}, + ]}, + "features": [ + {"id": "add_1", "atomic_id": "extrude_add_blind", "depends_on": [], + "params": {"distance_mm": 10}, "sketch_id": "s1"}, + {"id": "add_2", "atomic_id": "extrude_add_blind", "depends_on": ["add_1"], + "params": {"distance_mm": 10}, "sketch_id": "s2"}, + {"id": "hole_1", "atomic_id": "hole_wizard", "depends_on": ["add_2"], + "params": { + "hole_type": "简单直孔", "diameter_mm": 2.0, "depth_mm": 10.0, + "end_condition": {"type": "blind"}, + "positions": [{"mm": [0.0, 0.0, 10.0]}], + "host_face": {"kind": "face", "stable_id": "top", "source": "inferred_from_step", + "confidence": 1, "geometry": top_face["geometry"]}, + }}, + ], + } + rebuilt = rebuild_cdsl(with_hole, root / "hole.step") + + self.assertAlmostEqual(rebuilt["volume_mm3"], 2000 - math.pi * 10, places=5) + self.assertEqual(rebuilt["solid_count"], 2) + + def test_multi_body_collapse_reverts_to_flat_body_id(self) -> None: + """转换护栏：多体塌缩回单体后 body_id 恢复单体制（无 :index 后缀）。 + + 两个 box 合并为多体后，用大切除把第二个 box 整体切掉 → 主体恢复为 + 单个 Solid → register_body 应回到 body:{feature}（flat），后续 + selector 行为与普通单体零件完全一致。 + """ + base = _two_boss_doc() + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + with_cut = { + "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", + "part_id": "multi-body-collapse", "meta": {"unit": "mm"}, + "geometry": {"sketches": [ + {"id": "s1", "workplane": _workplane(origin=[0, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5])}, + {"id": "s2", "workplane": _workplane(origin=[20, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5])}, + {"id": "cut", "workplane": _workplane(origin=[14.5, 0, 0]), + "profile": _rectangle([0, -5.5], [11, 5.5])}, + ]}, + "features": [ + {"id": "add_1", "atomic_id": "extrude_add_blind", "depends_on": [], + "params": {"distance_mm": 10}, "sketch_id": "s1"}, + {"id": "add_2", "atomic_id": "extrude_add_blind", "depends_on": ["add_1"], + "params": {"distance_mm": 10}, "sketch_id": "s2"}, + {"id": "cut_2", "atomic_id": "extrude_cut_blind", "depends_on": ["add_2"], + "params": {"distance_mm": 10}, "sketch_id": "cut"}, + ], + } + rebuilt = rebuild_cdsl(with_cut, root / "collapse.step") + + self.assertAlmostEqual(rebuilt["volume_mm3"], 1000.0, places=5) + self.assertEqual(rebuilt["solid_count"], 1) + # 塌缩回单体：最后主体（body:cut_2）的面全部属于 flat 的 body_id + # （无 :index 成员后缀）——而不是像多体时那样带 body:cut_2:0/:1。 + # （registry 会保留历史快照供语义继承，因此只断言最后主体的记录。） + cut_faces = { + r["body_id"] for r in rebuilt["topology_records"] + if r["kind"] == "face" and r.get("body_id", "").startswith("body:cut_2") + } + self.assertEqual(cut_faces, {"body:cut_2"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_engine_selector_persistence.py b/backend/tests/test_engine_selector_persistence.py new file mode 100644 index 00000000..d3600470 --- /dev/null +++ b/backend/tests/test_engine_selector_persistence.py @@ -0,0 +1,221 @@ +"""#8 selector 持久性：上游 fillet 消费边之后的 selector 解析回归测试。 + +中文说明 +-------- +这个文件在测试什么（issue #8「Selector 持久性」的回归测试）： + + 1. 背景：selector 的 stable_id 与几何签名来自特征执行前的 B-rep。上游 + fillet/chamfer 会重建被选中边附近的拓扑： + - 被 fillet 直接选中的边会被双端圆角、拆分成"两条等距直段 + 两段 + 圆弧"，几何上不存在唯一"同一条边"的后继（本质歧义，CAD 中该边 + 也被视为已消费）； + - 与被圆角边共享端点的相邻边则只被"单端缩短"，方向不变、另一端 + 端点重合——它有一个明确的演化后继。 + 修复前：相邻边的 selector 在 fillet 之后 resolve 失败（not_found）， + 因为记录已迁移到新 body 且几何签名变了。 + 修复后（#8）：registry 记录"位置轨迹延续"的演化后继映射 + （old_record_id -> 漂移显著最小的唯一后继），resolve 的 stable_id + 精确匹配在 active body 过滤之前执行，命中过期记录时经演化后继解析 + 到 active body 内的新形态。 + + 2. 本测试套件把"被波及边可解析、被消费边保持保守"固定下来： + - 核心契约：fillet 圆角竖直边 A 后，引用相邻底面边 B 的 selector + 仍能 resolve，B 上的二次 fillet 重建成功、体积减少； + - 护栏：引用被完整消费的边 A 的 selector 保持 not_found（不把相邻 + 直段误匹配为 A 的延续，确定性优先）； + - 护栏：引用未被 fillet 波及的边的 selector 行为不变。 + + 3. sys.path 说明：把 backend 与 backend/engine 加入搜索路径，直接 import + cdsl_engine 包做端到端测试（与既有测试风格一致）。 + +函数功能一览 +------------ + _workplane(origin) 构造指定原点的 XY 平面草图工作平面。 + _rectangle(minimum, maximum) 构造 XY 平面内的矩形轮廓（2D 多边形）。 + _boss_doc() 10×10×10 单体拉伸文档（体积 1000）。 + _baseline_edges(...) 从基线重建里挑出 A/B/C 三条边并构造 selector。 + SelectorPersistenceTests 见各测试方法 docstring。 +""" + +from __future__ import annotations + +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "backend")) +sys.path.insert(0, str(ROOT / "backend" / "engine")) + +from cdsl_engine.runtime import rebuild_cdsl # noqa: E402 + + +# --------------------------------------------------------------------------- +# 测试夹具 +# --------------------------------------------------------------------------- + + +def _workplane(*, origin: list[float]) -> dict: + return {"origin_mm": origin, "x_dir": [1, 0, 0], "normal": [0, 0, 1]} + + +def _rectangle(minimum: list[float], maximum: list[float]) -> dict: + return {"type": "polygon", "vertices": [ + [minimum[0], minimum[1]], [maximum[0], minimum[1]], + [maximum[0], maximum[1]], [minimum[0], maximum[1]], + ]} + + +def _boss_doc() -> dict: + return { + "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", + "part_id": "selector-persist", "meta": {"unit": "mm"}, + "geometry": {"sketches": [{ + "id": "s1", "workplane": _workplane(origin=[0, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5]), + }]}, + "features": [ + {"id": "add_1", "atomic_id": "extrude_add_blind", "depends_on": [], + "params": {"distance_mm": 10}, "sketch_id": "s1"}, + ], + } + + +def _baseline_edges(): + """从基线重建里挑出测试用的三条边。 + + A = 竖直边 [-5,-5,0]→[-5,-5,10]（被 fillet 直接选中，将被完整消费）； + B = 底面边 [-5,-5,0]→[5,-5,0]（与 A 共享端点 [-5,-5,0]，被单端缩短）； + C = 竖直边 [5,5,0]→[5,5,10]（远离 A/B，完全不受 fillet 影响）。 + """ + with tempfile.TemporaryDirectory() as directory: + baseline = rebuild_cdsl(_boss_doc(), Path(directory) / "base.step") + + def _edge(match) -> dict: + return next( + item for item in baseline["topology_records"] + if item["kind"] == "edge" and match(item["geometry"]) + ) + + def _selector(record: dict) -> dict: + return {"kind": "edge", "stable_id": record["record_id"], + "source": "inferred_from_step", "confidence": 1, + "geometry": record["geometry"]} + + edge_a = _edge(lambda g: g.get("curve_type") == "line" + and g.get("start_mm") == [-5.0, -5.0, 0.0] + and g.get("end_mm") == [-5.0, -5.0, 10.0]) + edge_b = _edge(lambda g: g.get("curve_type") == "line" + and g.get("start_mm") == [-5.0, -5.0, 0.0] + and g.get("end_mm") == [5.0, -5.0, 0.0]) + edge_c = _edge(lambda g: g.get("curve_type") == "line" + and g.get("start_mm") == [5.0, 5.0, 0.0] + and g.get("end_mm") == [5.0, 5.0, 10.0]) + return _selector(edge_a), _selector(edge_b), _selector(edge_c) + + +# --------------------------------------------------------------------------- +# 测试套件 +# --------------------------------------------------------------------------- + + +class SelectorPersistenceTests(unittest.TestCase): + """#8 selector 持久性契约：被波及边可解析、被消费边保持保守。""" + + def test_consumed_adjacent_edge_selector_resolves_after_fillet(self) -> None: + """核心契约：fillet 圆角 A 后，引用相邻边 B 的 selector 仍可 resolve。 + + fillet_1 圆角竖直边 A（radius 2），把与其共享端点 [-5,-5,0] 的底面 + 边 B 单端缩短为 [-3,-5,0]→[5,-5,0]。fillet_2 用修复前捕获的 B + selector 再圆角一次（radius 1）： + - 修复前：B 记录已迁移到 body:fillet_1 且几何签名变化 → not_found， + 重建抛 RuntimeExecutionError； + - 修复后：演化后继解析到 B 的缩短形态 → 重建成功且体积减少。 + """ + selector_a, selector_b, _ = _baseline_edges() + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + baseline = rebuild_cdsl(_boss_doc(), root / "base.step") + with_fillets = { + "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", + "part_id": "selector-persist", "meta": {"unit": "mm"}, + "geometry": {"sketches": [{ + "id": "s1", "workplane": _workplane(origin=[0, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5]), + }]}, + "features": [ + {"id": "add_1", "atomic_id": "extrude_add_blind", "depends_on": [], + "params": {"distance_mm": 10}, "sketch_id": "s1"}, + {"id": "fillet_1", "atomic_id": "fillet", "depends_on": ["add_1"], + "params": {"radius_mm": 2}, "selectors": [selector_a]}, + {"id": "fillet_2", "atomic_id": "fillet", "depends_on": ["fillet_1"], + "params": {"radius_mm": 1}, "selectors": [selector_b]}, + ], + } + rebuilt = rebuild_cdsl(with_fillets, root / "two-fillets.step") + self.assertLess(rebuilt["volume_mm3"], baseline["volume_mm3"]) + self.assertGreater(rebuilt["volume_mm3"], 900.0) + + def test_fully_consumed_edge_selector_stays_conservative(self) -> None: + """护栏：被完整消费的边 A 的 selector 保持 not_found（确定性优先）。 + + fillet_1 圆角 A 后，A 被拆分成两条等距直段 + 两段圆弧，几何上没有 + 唯一后继（两条直段到原边的漂移并列）。此场景下 registry 不登记演化 + 映射，二次引用 A 应明确失败，而不是把相邻直段误匹配成 A 的延续。 + """ + selector_a, _, _ = _baseline_edges() + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + with_second = { + "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", + "part_id": "selector-persist", "meta": {"unit": "mm"}, + "geometry": {"sketches": [{ + "id": "s1", "workplane": _workplane(origin=[0, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5]), + }]}, + "features": [ + {"id": "add_1", "atomic_id": "extrude_add_blind", "depends_on": [], + "params": {"distance_mm": 10}, "sketch_id": "s1"}, + {"id": "fillet_1", "atomic_id": "fillet", "depends_on": ["add_1"], + "params": {"radius_mm": 2}, "selectors": [selector_a]}, + {"id": "fillet_2", "atomic_id": "fillet", "depends_on": ["fillet_1"], + "params": {"radius_mm": 1}, "selectors": [selector_a]}, + ], + } + with self.assertRaises(Exception) as caught: + rebuild_cdsl(with_second, root / "second.step") + message = str(caught.exception) + self.assertIn("No runtime topology record satisfies the selector", message) + + def test_unaffected_edge_selector_resolves_after_fillet(self) -> None: + """护栏：远离 fillet 的边 C 的 selector 行为不受影响。 + + fillet_1 圆角 A 后，引用 C（[5,5,0]→[5,5,10]，几何完全不变）的 + selector 通过几何打分照常 resolve，二次 fillet 重建成功。 + """ + selector_a, _, selector_c = _baseline_edges() + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + with_fillets = { + "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", + "part_id": "selector-persist", "meta": {"unit": "mm"}, + "geometry": {"sketches": [{ + "id": "s1", "workplane": _workplane(origin=[0, 0, 0]), + "profile": _rectangle([-5, -5], [5, 5]), + }]}, + "features": [ + {"id": "add_1", "atomic_id": "extrude_add_blind", "depends_on": [], + "params": {"distance_mm": 10}, "sketch_id": "s1"}, + {"id": "fillet_1", "atomic_id": "fillet", "depends_on": ["add_1"], + "params": {"radius_mm": 2}, "selectors": [selector_a]}, + {"id": "fillet_2", "atomic_id": "fillet", "depends_on": ["fillet_1"], + "params": {"radius_mm": 1}, "selectors": [selector_c]}, + ], + } + rebuilt = rebuild_cdsl(with_fillets, root / "with-c.step") + self.assertLess(rebuilt["volume_mm3"], 1000.0) + + +if __name__ == "__main__": + unittest.main()