Sync from forma-3d-review@b6d4687f5d0f2e5303758c97095ea7e38e740723

.dockerignore

@@ -0,0 +1,14 @@
+.venv/
+.pytest_cache/
+__pycache__/
+*.pyc
+.git
+.github
+jobs/
+tests/
+docs/
+*.md
+!README.md
+.env
+.env.*
+.python-version

.env.example

@@ -0,0 +1,7 @@
+# Comma-separated list of allowed origins. Use "*" only for local dev.
+ALLOWED_ORIGINS=http://localhost:5173
+
+# Override the directory that `/api/samples` scans for example STEP files.
+# Defaults to `<repo>/data`. In FastAPI Cloud you can leave this unset if the
+# data folder is included in the deployed source tree.
+# SAMPLES_DIR=/app/data

.gitattributes

@@ -33,3 +33,7 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+data/Print1.png filter=lfs diff=lfs merge=lfs -text
+data/Print2.png filter=lfs diff=lfs merge=lfs -text
+data/Print3.png filter=lfs diff=lfs merge=lfs -text
+data/turbine_interior.step filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -0,0 +1,35 @@
+FROM python:3.11-slim
+
+# System libs required by cadquery-ocp (OpenCascade) and open3d.
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends \
+        libgl1 \
+        libglib2.0-0 \
+        libxrender1 \
+        libxext6 \
+        libsm6 \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
+
+ENV UV_LINK_MODE=copy \
+    UV_COMPILE_BYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONDONTWRITEBYTECODE=1
+
+WORKDIR /app
+
+# Install dependencies first for layer caching.
+COPY pyproject.toml uv.lock ./
+RUN uv sync --frozen --no-install-project --no-dev
+
+# Copy the application source.
+COPY . .
+
+# Install the project itself (editable) so backend/data/ is resolvable.
+RUN uv sync --frozen --no-dev
+
+ENV PORT=8080
+EXPOSE 8080
+
+CMD ["sh", "-c", "uv run uvicorn src.api.main:app --host 0.0.0.0 --port ${PORT}"]

README.md

@@ -1,11 +1,15 @@
 ---
-title: Forma 3d Review Api
-emoji: 📉
-colorFrom: gray
-colorTo: green
+title: Forma 3D Review API
+emoji: 🔧
+colorFrom: blue
+colorTo: indigo
 sdk: docker
+app_port: 8080
 pinned: false
-license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Forma 3D Review API
+
+Backend for the 3D CAD assembly review tool. Built with FastAPI + cadquery-ocp.
+Source of truth lives at <https://github.com/DShomin/forma-3d-review>; this
+Space is synced from the `backend/` subtree on every push to `main`.

config/kmvss_rules.yaml

@@ -0,0 +1,40 @@
+rules:
+  - id: "turbine-nacelle-diameter"
+    name: "Nacelle Maximum Diameter"
+    description: "Nacelle housing diameter must not exceed specified limit"
+    type: bounding_box
+    params:
+      target_name_pattern: "nacelle|housing|cowl"
+      axis: "x"
+      max_mm: 5000.0
+    severity: error
+
+  - id: "blade-clearance"
+    name: "Blade Tip Clearance"
+    description: "Minimum clearance between blade tips and nacelle inner wall"
+    type: clearance
+    params:
+      part_a_pattern: "blade|rotor"
+      part_b_pattern: "nacelle|housing|shroud"
+      min_clearance_mm: 10.0
+    severity: error
+
+  - id: "hub-height"
+    name: "Hub Assembly Height"
+    description: "Hub assembly height within specification"
+    type: bounding_box
+    params:
+      target_name_pattern: "hub"
+      axis: "z"
+      max_mm: 3000.0
+    severity: warning
+
+  - id: "shaft-alignment"
+    name: "Main Shaft Concentricity"
+    description: "Main shaft center deviation from assembly center"
+    type: bounding_box
+    params:
+      target_name_pattern: "shaft|axle"
+      axis: "y"
+      max_mm: 2000.0
+    severity: warning

config/settings.py

@@ -0,0 +1,62 @@
+"""Configuration settings for the CAD review tool."""
+from pathlib import Path
+
+from pydantic import Field, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+_BACKEND_ROOT = Path(__file__).resolve().parents[1]
+_REPO_ROOT = _BACKEND_ROOT.parent
+_SAMPLES_CANDIDATES = (_BACKEND_ROOT / "data", _REPO_ROOT / "data")
+
+
+def _default_samples_dir() -> Path:
+    for candidate in _SAMPLES_CANDIDATES:
+        if candidate.exists():
+            return candidate
+    return _SAMPLES_CANDIDATES[0]
+
+
+class TessellationSettings(BaseSettings):
+    deflection_factor: float = Field(default=0.1)
+    min_deflection_mm: float = Field(default=0.001)
+    max_deflection_mm: float = Field(default=0.1)
+    angular_deflection_rad: float = Field(default=0.5)
+
+
+class SamplingSettings(BaseSettings):
+    num_samples: int = Field(default=100_000)
+    seed: int = Field(default=42)
+
+
+class ToleranceSettings(BaseSettings):
+    g0_position_mm: float = Field(default=0.05)
+    g1_tangent_deg: float = Field(default=0.3)
+    g2_curvature_pct: float = Field(default=3.0)
+    gap_tolerance_mm: float = Field(default=0.05)
+    overlap_tolerance_mm: float = Field(default=0.05)
+
+
+class SplitterSettings(BaseSettings):
+    exterior_patterns: list[str] = Field(
+        default=["inlet front", "inlet back"],
+        description="Part name patterns for exterior group (case-insensitive substring match)",
+    )
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+
+    tessellation: TessellationSettings = Field(default_factory=TessellationSettings)
+    sampling: SamplingSettings = Field(default_factory=SamplingSettings)
+    tolerance: ToleranceSettings = Field(default_factory=ToleranceSettings)
+    splitter: SplitterSettings = Field(default_factory=SplitterSettings)
+
+    samples_dir: Path = Field(default_factory=_default_samples_dir)
+    allowed_origins: list[str] = Field(default_factory=lambda: ["*"])
+
+    @field_validator("allowed_origins", mode="before")
+    @classmethod
+    def _split_origins(cls, v: object) -> object:
+        if isinstance(v, str):
+            return [s.strip() for s in v.split(",") if s.strip()]
+        return v

data/turbine_exterior.names.json

@@ -0,0 +1,4 @@
+[
+  "Inlet Front v9",
+  "Inlet Back v3"
+]

data/turbine_interior.names.json

@@ -0,0 +1,192 @@
+[
+  "F1 Fan v6",
+  "Center Bolt v4",
+  "Assembled Shaft v12",
+  "skf_bearing_nup_2306_ecml_2 v2",
+  "Shaft Front v17",
+  "Parafuso sextavado conformado ANSI B18.2.3.2M - M10x1.5 x 16 A\u00e7o grau 2 Liso v1",
+  "Shaft Mid v3",
+  "Shaft Back v5",
+  "skf_bearing_nup_2306_ecml_2 v2",
+  "skf_bearing_nup_2306_ecml_2_02",
+  "skf_bearing_nup_2306_ecml_2_03",
+  "skf_bearing_nup_2306_ecml_2_01",
+  "Ring F1 v5",
+  "Ring F2 v6",
+  "CF1 v6",
+  "CR1 v7",
+  "Compressor Shell v6",
+  "CF2 v5",
+  "CR2 v7",
+  "CF3 v5",
+  "CR3 v2",
+  "CF4 v3",
+  "CR4 v2",
+  "CC Shell v6",
+  "CC Fan v3",
+  "HP Shell v11",
+  "HP Fan v5",
+  "LP Fan1 v3",
+  "LP Fan2 v2",
+  "LP Fan3 v2",
+  "Exhaust Fan v4",
+  "Exhaust Bolt v4",
+  "Exhaust Shell Front v17",
+  "Exhaust Shell Back v3",
+  "Part_37",
+  "Part_38",
+  "Part_39",
+  "Part_40",
+  "Part_41",
+  "Part_42",
+  "Part_43",
+  "Part_44",
+  "Part_45",
+  "Part_46",
+  "Part_47",
+  "Part_48",
+  "Part_49",
+  "Part_50",
+  "Part_51",
+  "Part_52",
+  "Part_53",
+  "Part_54",
+  "Part_55",
+  "Part_56",
+  "Part_57",
+  "Part_58",
+  "Part_59",
+  "Part_60",
+  "Part_61",
+  "Part_62",
+  "Part_63",
+  "Part_64",
+  "Part_65",
+  "Part_66",
+  "Part_67",
+  "Part_68",
+  "Part_69",
+  "Part_70",
+  "Part_71",
+  "Part_72",
+  "Part_73",
+  "Part_74",
+  "Part_75",
+  "Part_76",
+  "Part_77",
+  "Part_78",
+  "Part_79",
+  "Part_80",
+  "Part_81",
+  "Part_82",
+  "Part_83",
+  "Part_84",
+  "Part_85",
+  "Part_86",
+  "Part_87",
+  "Part_88",
+  "Part_89",
+  "Part_90",
+  "Part_91",
+  "Part_92",
+  "Part_93",
+  "Part_94",
+  "Part_95",
+  "Part_96",
+  "Part_97",
+  "Part_98",
+  "Part_99",
+  "Part_100",
+  "Part_101",
+  "Part_102",
+  "Part_103",
+  "Part_104",
+  "Part_105",
+  "Part_106",
+  "Part_107",
+  "Part_108",
+  "Part_109",
+  "Part_110",
+  "Part_111",
+  "Part_112",
+  "Part_113",
+  "Part_114",
+  "Part_115",
+  "Part_116",
+  "Part_117",
+  "Part_118",
+  "Part_119",
+  "Part_120",
+  "Part_121",
+  "Part_122",
+  "Part_123",
+  "Part_124",
+  "Part_125",
+  "Part_126",
+  "Part_127",
+  "Part_128",
+  "Part_129",
+  "Part_130",
+  "Part_131",
+  "Part_132",
+  "Part_133",
+  "Part_134",
+  "Part_135",
+  "Part_136",
+  "Part_137",
+  "Part_138",
+  "Part_139",
+  "Part_140",
+  "Part_141",
+  "Part_142",
+  "Part_143",
+  "Part_144",
+  "Part_145",
+  "Part_146",
+  "Part_147",
+  "Part_148",
+  "Part_149",
+  "Part_150",
+  "Part_151",
+  "Part_152",
+  "Part_153",
+  "Part_154",
+  "Part_155",
+  "Part_156",
+  "Part_157",
+  "Part_158",
+  "Part_159",
+  "Part_160",
+  "Part_161",
+  "Part_162",
+  "Part_163",
+  "Part_164",
+  "Part_165",
+  "Part_166",
+  "Part_167",
+  "Part_168",
+  "Part_169",
+  "Part_170",
+  "Part_171",
+  "Part_172",
+  "Part_173",
+  "Part_174",
+  "Part_175",
+  "Part_176",
+  "Part_177",
+  "Part_178",
+  "Part_179",
+  "Part_180",
+  "Part_181",
+  "Part_182",
+  "Part_183",
+  "Part_184",
+  "Part_185",
+  "Part_186",
+  "Part_187",
+  "Part_188",
+  "Part_189",
+  "Part_190",
+  "Part_191",
+  "Part_192"
+]

data/turbine_interior.step

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:46f0fc9d1de95c28e756ec57e2b1284bcd892a6605908485e80d2be8245408d6
+size 52856628

main.py

@@ -0,0 +1,8 @@
+"""FastAPI Cloud entry point.
+
+FastAPI Cloud's default `fastapi run` looks for an `app` object in
+`main.py` at the project root. Re-export it from `src.api.main`.
+"""
+from src.api.main import app
+
+__all__ = ["app"]

pyproject.toml

@@ -0,0 +1,37 @@
+[project]
+name = "cad-review"
+version = "0.1.0"
+description = "3D CAD assembly review tool with exterior/interior splitting and compliance checking"
+requires-python = ">=3.11,<3.12"
+dependencies = [
+    "cadquery-ocp>=7.7",
+    "numpy>=1.26",
+    "open3d>=0.18",
+    "pydantic-settings>=2.0",
+    "fastapi[standard]>=0.115",
+    "uvicorn[standard]>=0.30",
+    "python-multipart>=0.0.9",
+    "pygltflib>=1.16",
+    "pyyaml>=6.0",
+    "rich>=13.0",
+]
+
+[project.scripts]
+cad-review-api = "src.api.main:run_server"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src", "config"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["."]
+
+[dependency-groups]
+dev = [
+    "pytest>=9.0.2",
+    "httpx>=0.27",
+]

src/api/job_manager.py

@@ -0,0 +1,274 @@
+"""Asynchronous job management for the review pipeline."""
+from __future__ import annotations
+
+import asyncio
+import logging
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from config.settings import Settings
+from src.api.mesh_export import mesh_to_glb
+from src.geometry.tessellator import tessellate_faces
+from src.loader.assembly_tree import AssemblyNode, extract_assembly_tree, extract_assembly_from_shape
+from src.loader.step_loader import load_step
+
+logger = logging.getLogger(__name__)
+
+PIPELINE_STEPS = [
+    "loading",
+    "assembly_tree",
+    "tessellation",
+    "classification",
+    "mesh_export",
+]
+
+
+@dataclass
+class PartMeshData:
+    """Mesh data for a single part."""
+    part_id: str
+    vertices: np.ndarray
+    triangles: np.ndarray
+    glb: bytes
+
+
+@dataclass
+class GroupInfo:
+    """Info about a part group (exterior / interior)."""
+    name: str
+    part_ids: list[str]
+    part_names: list[str]
+
+
+@dataclass
+class JobData:
+    """Stores all data associated with a review job."""
+
+    job_id: str
+    exterior_path: Path
+    interior_path: Path
+    status: str = "pending"  # pending, running, completed, failed
+    error: str | None = None
+
+    # Step tracking
+    steps: dict[str, str] = field(default_factory=dict)
+
+    # Results
+    assembly_tree: AssemblyNode | None = None
+    part_meshes: dict[str, PartMeshData] = field(default_factory=dict)
+    group_meshes: dict[str, PartMeshData] = field(default_factory=dict)
+    groups: dict[str, GroupInfo] = field(default_factory=dict)
+    exterior_step_info: Any = None
+    interior_step_info: Any = None
+
+    # SSE subscribers
+    _listeners: list[asyncio.Queue] = field(default_factory=list)
+    _loop: asyncio.AbstractEventLoop | None = None
+
+    def __post_init__(self) -> None:
+        for step in PIPELINE_STEPS:
+            self.steps[step] = "pending"
+
+
+class JobManager:
+    """Manages review jobs with thread-based execution and SSE streaming."""
+
+    def __init__(self, upload_dir: Path | None = None) -> None:
+        self._jobs: dict[str, JobData] = {}
+        self._executor = ThreadPoolExecutor(max_workers=2)
+        self._upload_dir = upload_dir or Path("/tmp/cad-review-uploads")
+        self._upload_dir.mkdir(parents=True, exist_ok=True)
+        self._settings = Settings()
+
+    @property
+    def upload_dir(self) -> Path:
+        return self._upload_dir
+
+    @property
+    def settings(self) -> Settings:
+        return self._settings
+
+    def create_job(self, exterior_path: Path, interior_path: Path) -> JobData:
+        """Create a new review job with exterior and interior STEP files."""
+        job_id = uuid.uuid4().hex[:12]
+        job = JobData(job_id=job_id, exterior_path=exterior_path, interior_path=interior_path)
+        self._jobs[job_id] = job
+        return job
+
+    def get_job(self, job_id: str) -> JobData | None:
+        return self._jobs.get(job_id)
+
+    def subscribe(self, job: JobData) -> asyncio.Queue:
+        queue: asyncio.Queue = asyncio.Queue()
+        job._listeners.append(queue)
+        return queue
+
+    def unsubscribe(self, job: JobData, queue: asyncio.Queue) -> None:
+        if queue in job._listeners:
+            job._listeners.remove(queue)
+
+    def _emit(self, job: JobData, event: dict) -> None:
+        loop = job._loop
+        if loop is None:
+            return
+        for q in job._listeners:
+            loop.call_soon_threadsafe(q.put_nowait, event)
+
+    def _emit_step(self, job: JobData, step: str, status: str, pct: int = 0, **kwargs) -> None:
+        job.steps[step] = status
+        event = {"step": step, "status": status, "progress_pct": pct, **kwargs}
+        self._emit(job, event)
+
+    def start_job(self, job: JobData, loop: asyncio.AbstractEventLoop) -> None:
+        """Start job execution in a background thread."""
+        job.status = "running"
+        job._loop = loop
+        self._executor.submit(self._run_pipeline, job)
+
+    def _extract_tree(self, step_info, id_prefix: str = "") -> AssemblyNode:
+        """Extract assembly tree from a loaded STEP file, trying XDE first."""
+        tree = None
+        if step_info.shape_tool is not None:
+            try:
+                tree = extract_assembly_tree(step_info.shape_tool)
+            except Exception as e:
+                logger.warning("XDE tree extraction failed: %s, falling back to shape topology", e)
+        if tree is None:
+            tree = extract_assembly_from_shape(
+                step_info.shape, step_info.path.stem,
+                reader=step_info.reader, doc=step_info.doc,
+                step_path=step_info.path,
+                id_prefix=id_prefix,
+            )
+        return tree
+
+    def _run_pipeline(self, job: JobData) -> None:
+        """Execute the review pipeline in a worker thread."""
+        try:
+            settings = self._settings
+
+            # Step 1: Load both STEP files
+            self._emit_step(job, "loading", "running")
+            exterior_info = load_step(job.exterior_path)
+            job.exterior_step_info = exterior_info
+            self._emit_step(job, "loading", "running", 50)
+            interior_info = load_step(job.interior_path)
+            job.interior_step_info = interior_info
+            self._emit_step(job, "loading", "completed", 100)
+
+            # Step 2: Extract assembly trees and build combined root
+            self._emit_step(job, "assembly_tree", "running")
+            exterior_tree = self._extract_tree(exterior_info, id_prefix="ext_")
+            exterior_tree.name = f"Exterior - {exterior_tree.name}"
+            interior_tree = self._extract_tree(interior_info, id_prefix="int_")
+            interior_tree.name = f"Interior - {interior_tree.name}"
+
+            # Tag all leaves with their classification
+            for leaf in exterior_tree.iter_leaves():
+                leaf.classification = "exterior"
+            for leaf in interior_tree.iter_leaves():
+                leaf.classification = "interior"
+
+            # Build combined root
+            tree = AssemblyNode(
+                id="root",
+                name="Combined Assembly",
+                is_assembly=True,
+                children=[exterior_tree, interior_tree],
+            )
+            job.assembly_tree = tree
+            self._emit_step(job, "assembly_tree", "completed", 100)
+
+            # Step 3: Tessellate each leaf part
+            self._emit_step(job, "tessellation", "running")
+            leaves = list(tree.iter_leaves())
+            total_leaves = len(leaves)
+
+            for idx, leaf in enumerate(leaves):
+                if leaf.shape is None or leaf.shape.IsNull():
+                    continue
+                try:
+                    verts, tris = tessellate_faces(
+                        leaf.shape, settings.tessellation,
+                        target_tolerance_mm=settings.tolerance.g0_position_mm,
+                    )
+                    leaf.num_faces = len(tris)
+                    pct = int((idx + 1) / total_leaves * 100)
+                    self._emit_step(job, "tessellation", "running", pct)
+
+                    # Store mesh data
+                    job.part_meshes[leaf.id] = PartMeshData(
+                        part_id=leaf.id,
+                        vertices=verts,
+                        triangles=tris,
+                        glb=b"",  # Generate later
+                    )
+                except Exception as e:
+                    logger.warning("Failed to tessellate part '%s': %s", leaf.name, e)
+
+            self._emit_step(job, "tessellation", "completed", 100)
+
+            # Step 4: Classification (by file origin - already tagged above)
+            self._emit_step(job, "classification", "running")
+            for group_name, subtree in [("exterior", exterior_tree), ("interior", interior_tree)]:
+                group_leaves = list(subtree.iter_leaves())
+                job.groups[group_name] = GroupInfo(
+                    name=group_name,
+                    part_ids=[leaf.id for leaf in group_leaves if leaf.shape is not None],
+                    part_names=[leaf.name for leaf in group_leaves if leaf.shape is not None],
+                )
+            self._emit_step(job, "classification", "completed", 100)
+
+            # Step 5: Export meshes to GLB (individual + combined groups)
+            self._emit_step(job, "mesh_export", "running")
+            for part_id, mesh_data in job.part_meshes.items():
+                try:
+                    mesh_data.glb = mesh_to_glb(mesh_data.vertices, mesh_data.triangles)
+                except Exception as e:
+                    logger.warning("Failed to export GLB for part %s: %s", part_id, e)
+
+            # Create combined group meshes
+            for group_name, group_info in job.groups.items():
+                group_verts_list = []
+                group_tris_list = []
+                vert_offset = 0
+                for pid in group_info.part_ids:
+                    md = job.part_meshes.get(pid)
+                    if md is None:
+                        continue
+                    group_verts_list.append(md.vertices)
+                    group_tris_list.append(md.triangles + vert_offset)
+                    vert_offset += len(md.vertices)
+                if group_verts_list:
+                    combined_verts = np.concatenate(group_verts_list, axis=0)
+                    combined_tris = np.concatenate(group_tris_list, axis=0)
+                    try:
+                        combined_glb = mesh_to_glb(combined_verts, combined_tris)
+                        job.group_meshes[group_name] = PartMeshData(
+                            part_id=f"{group_name}_combined",
+                            vertices=combined_verts,
+                            triangles=combined_tris,
+                            glb=combined_glb,
+                        )
+                    except Exception as e:
+                        logger.warning("Failed to create combined GLB for %s: %s", group_name, e)
+
+            self._emit_step(job, "mesh_export", "completed", 100, overall_status="completed")
+
+            job.status = "completed"
+
+        except Exception as e:
+            logger.exception("Job %s failed", job.job_id)
+            job.status = "failed"
+            job.error = str(e)
+            self._emit(job, {
+                "step": "error",
+                "status": "failed",
+                "error": str(e),
+                "overall_status": "failed",
+            })

src/api/main.py

@@ -0,0 +1,41 @@
+"""FastAPI application entry point."""
+from __future__ import annotations
+
+import logging
+import uvicorn
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from config.settings import Settings
+from src.api.router import router
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")
+
+
+def create_app() -> FastAPI:
+    app = FastAPI(
+        title="CAD Review API",
+        description="3D CAD assembly review with exterior/interior splitting and compliance checking",
+        version="0.1.0",
+    )
+
+    settings = Settings()
+    allow_credentials = settings.allowed_origins != ["*"]
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=settings.allowed_origins,
+        allow_credentials=allow_credentials,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    app.include_router(router)
+    return app
+
+
+app = create_app()
+
+
+def run_server() -> None:
+    uvicorn.run("src.api.main:app", host="0.0.0.0", port=8000, reload=True)

src/api/mesh_export.py

@@ -0,0 +1,137 @@
+"""Export mesh data (vertices + triangles) to glTF binary (.glb) format."""
+
+from __future__ import annotations
+
+import numpy as np
+import pygltflib
+
+
+def _compute_normals(vertices: np.ndarray, triangles: np.ndarray) -> np.ndarray:
+    """Compute smooth per-vertex normals by averaging face normals."""
+    normals = np.zeros_like(vertices, dtype=np.float32)
+
+    v0 = vertices[triangles[:, 0]]
+    v1 = vertices[triangles[:, 1]]
+    v2 = vertices[triangles[:, 2]]
+
+    face_normals = np.cross(v1 - v0, v2 - v0)
+
+    # Accumulate face normals to each vertex
+    for i in range(3):
+        np.add.at(normals, triangles[:, i], face_normals)
+
+    # Normalize
+    lengths = np.linalg.norm(normals, axis=1, keepdims=True)
+    lengths = np.maximum(lengths, 1e-10)
+    normals /= lengths
+
+    return normals.astype(np.float32)
+
+
+def mesh_to_glb(vertices: np.ndarray, triangles: np.ndarray) -> bytes:
+    """Convert vertices and triangles to a binary glTF (.glb) buffer.
+
+    Includes computed vertex normals for proper lighting/shading.
+
+    Args:
+        vertices: Mesh vertices with shape (N, 3), float64.
+        triangles: Triangle indices with shape (M, 3), int32.
+
+    Returns:
+        Bytes of the .glb file.
+    """
+    vertices_f32 = vertices.astype(np.float32)
+    triangles_u32 = triangles.astype(np.uint32)
+    normals_f32 = _compute_normals(vertices_f32, triangles_u32)
+
+    # Byte buffers
+    vert_bytes = vertices_f32.tobytes()
+    norm_bytes = normals_f32.tobytes()
+    tri_bytes = triangles_u32.tobytes()
+
+    # Align each buffer to 4 bytes
+    def pad4(b: bytes) -> bytes:
+        pad = (4 - len(b) % 4) % 4
+        return b + b"\x00" * pad
+
+    vert_padded = pad4(vert_bytes)
+    norm_padded = pad4(norm_bytes)
+
+    blob = vert_padded + norm_padded + tri_bytes
+    total_bytes = len(blob)
+
+    norm_offset = len(vert_padded)
+    tri_offset = norm_offset + len(norm_padded)
+
+    v_min = vertices_f32.min(axis=0).tolist()
+    v_max = vertices_f32.max(axis=0).tolist()
+
+    gltf = pygltflib.GLTF2(
+        scene=0,
+        scenes=[pygltflib.Scene(nodes=[0])],
+        nodes=[pygltflib.Node(mesh=0)],
+        meshes=[
+            pygltflib.Mesh(
+                primitives=[
+                    pygltflib.Primitive(
+                        attributes=pygltflib.Attributes(POSITION=0, NORMAL=1),
+                        indices=2,
+                    )
+                ]
+            )
+        ],
+        accessors=[
+            # Accessor 0: vertex positions
+            pygltflib.Accessor(
+                bufferView=0,
+                componentType=pygltflib.FLOAT,
+                count=len(vertices_f32),
+                type=pygltflib.VEC3,
+                max=v_max,
+                min=v_min,
+            ),
+            # Accessor 1: vertex normals
+            pygltflib.Accessor(
+                bufferView=1,
+                componentType=pygltflib.FLOAT,
+                count=len(normals_f32),
+                type=pygltflib.VEC3,
+            ),
+            # Accessor 2: triangle indices
+            pygltflib.Accessor(
+                bufferView=2,
+                componentType=pygltflib.UNSIGNED_INT,
+                count=triangles_u32.size,
+                type=pygltflib.SCALAR,
+                max=[int(triangles_u32.max())],
+                min=[int(triangles_u32.min())],
+            ),
+        ],
+        bufferViews=[
+            # BufferView 0: positions
+            pygltflib.BufferView(
+                buffer=0,
+                byteOffset=0,
+                byteLength=len(vert_bytes),
+                target=pygltflib.ARRAY_BUFFER,
+            ),
+            # BufferView 1: normals
+            pygltflib.BufferView(
+                buffer=0,
+                byteOffset=norm_offset,
+                byteLength=len(norm_bytes),
+                target=pygltflib.ARRAY_BUFFER,
+            ),
+            # BufferView 2: indices
+            pygltflib.BufferView(
+                buffer=0,
+                byteOffset=tri_offset,
+                byteLength=len(tri_bytes),
+                target=pygltflib.ELEMENT_ARRAY_BUFFER,
+            ),
+        ],
+        buffers=[pygltflib.Buffer(byteLength=total_bytes)],
+    )
+
+    gltf.set_binary_blob(blob)
+    return b"".join(gltf.save_to_bytes())

src/api/router.py

@@ -0,0 +1,378 @@
+"""API endpoint definitions for the CAD review tool."""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from pathlib import Path
+
+from fastapi import APIRouter, HTTPException, UploadFile, File
+from fastapi.responses import FileResponse, Response, StreamingResponse
+
+from src.api.job_manager import JobManager, PIPELINE_STEPS
+from src.api.samples import discover_samples, get_sample
+from src.api.schemas import (
+    AssemblyNodeResponse,
+    AssemblyTreeResponse,
+    ComplianceResponse,
+    ComplianceResultResponse,
+    ComplianceRuleResponse,
+    DistanceMeasurementResponse,
+    DistanceRequest,
+    GroupInfoResponse,
+    JobStatus,
+    ProximityPairResponse,
+    ProximityResponse,
+    SampleFileResponse,
+    SamplePairResponse,
+    StepProgress,
+    UploadResponse,
+)
+from src.compliance.kmvss_checker import check_compliance
+from src.compliance.rule_loader import load_rules
+from src.geometry.measurement import measure_distance, scan_proximity
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api")
+
+# Singleton job manager
+_manager: JobManager | None = None
+
+
+def get_manager() -> JobManager:
+    global _manager
+    if _manager is None:
+        _manager = JobManager()
+    return _manager
+
+
+def _node_to_response(node_dict: dict) -> AssemblyNodeResponse:
+    """Convert assembly node dict to response model."""
+    children = [_node_to_response(c) for c in node_dict.get("children", [])]
+    return AssemblyNodeResponse(
+        id=node_dict["id"],
+        name=node_dict["name"],
+        is_assembly=node_dict["is_assembly"],
+        is_leaf=node_dict["is_leaf"],
+        classification=node_dict.get("classification", "unknown"),
+        num_faces=node_dict.get("num_faces", 0),
+        num_solids=node_dict.get("num_solids", 0),
+        bounding_box=node_dict.get("bounding_box"),
+        children=children,
+    )
+
+
+@router.post("/upload", response_model=UploadResponse, status_code=201)
+async def upload_files(
+    exterior: UploadFile = File(..., description="Exterior (design) STEP file"),
+    interior: UploadFile = File(..., description="Interior (engineering) STEP file"),
+) -> UploadResponse:
+    """Upload exterior and interior STEP files and create a review job."""
+    manager = get_manager()
+
+    for label, f in [("Exterior", exterior), ("Interior", interior)]:
+        if f.filename and not f.filename.lower().endswith((".stp", ".step")):
+            raise HTTPException(400, f"Invalid {label} file type: {f.filename}. Expected .stp or .step")
+
+    exterior_path = manager.upload_dir / (exterior.filename or "exterior.step")
+    exterior_content = await exterior.read()
+    exterior_path.write_bytes(exterior_content)
+
+    interior_path = manager.upload_dir / (interior.filename or "interior.step")
+    interior_content = await interior.read()
+    interior_path.write_bytes(interior_content)
+
+    job = manager.create_job(exterior_path, interior_path)
+
+    loop = asyncio.get_event_loop()
+    manager.start_job(job, loop)
+
+    return UploadResponse(
+        job_id=job.job_id,
+        status=job.status,
+        exterior_filename=exterior.filename or "exterior.step",
+        interior_filename=interior.filename or "interior.step",
+    )
+
+
+@router.get("/samples", response_model=list[SamplePairResponse])
+async def list_samples() -> list[SamplePairResponse]:
+    """List available sample STEP pairs discoverable in the samples directory."""
+    manager = get_manager()
+    pairs = discover_samples(manager.settings.samples_dir)
+    return [
+        SamplePairResponse(
+            id=p.id,
+            name=p.name,
+            exterior=SampleFileResponse(filename=p.exterior.filename, size=p.exterior.size),
+            interior=SampleFileResponse(filename=p.interior.filename, size=p.interior.size),
+        )
+        for p in pairs
+    ]
+
+
+@router.get("/samples/{sample_id}/files/{kind}")
+async def download_sample_file(sample_id: str, kind: str) -> FileResponse:
+    """Download a sample STEP file (kind: 'exterior' or 'interior')."""
+    if kind not in ("exterior", "interior"):
+        raise HTTPException(400, f"Invalid kind: {kind}. Must be 'exterior' or 'interior'.")
+
+    manager = get_manager()
+    pair = get_sample(manager.settings.samples_dir, sample_id)
+    if pair is None:
+        raise HTTPException(404, f"Sample not found: {sample_id}")
+
+    file = pair.exterior if kind == "exterior" else pair.interior
+    return FileResponse(
+        path=file.path,
+        filename=file.filename,
+        media_type="application/octet-stream",
+    )
+
+
+@router.post("/samples/{sample_id}/load", response_model=UploadResponse, status_code=201)
+async def load_sample(sample_id: str) -> UploadResponse:
+    """Start a review job using a server-side sample pair (no upload required)."""
+    manager = get_manager()
+    pair = get_sample(manager.settings.samples_dir, sample_id)
+    if pair is None:
+        raise HTTPException(404, f"Sample not found: {sample_id}")
+
+    job = manager.create_job(pair.exterior.path, pair.interior.path)
+
+    loop = asyncio.get_event_loop()
+    manager.start_job(job, loop)
+
+    return UploadResponse(
+        job_id=job.job_id,
+        status=job.status,
+        exterior_filename=pair.exterior.filename,
+        interior_filename=pair.interior.filename,
+    )
+
+
+@router.get("/jobs/{job_id}", response_model=JobStatus)
+async def get_job_status(job_id: str) -> JobStatus:
+    """Get the current status of a job."""
+    manager = get_manager()
+    job = manager.get_job(job_id)
+    if job is None:
+        raise HTTPException(404, f"Job not found: {job_id}")
+
+    steps = [
+        StepProgress(step=name, status=job.steps.get(name, "pending"))
+        for name in PIPELINE_STEPS
+    ]
+
+    return JobStatus(
+        job_id=job.job_id,
+        status=job.status,
+        steps=steps,
+        error=job.error,
+    )
+
+
+@router.get("/jobs/{job_id}/events")
+async def job_events(job_id: str) -> StreamingResponse:
+    """SSE stream of pipeline progress events."""
+    manager = get_manager()
+    job = manager.get_job(job_id)
+    if job is None:
+        raise HTTPException(404, f"Job not found: {job_id}")
+
+    queue = manager.subscribe(job)
+
+    async def event_stream():
+        try:
+            while True:
+                event = await asyncio.wait_for(queue.get(), timeout=300)
+                yield f"data: {json.dumps(event)}\n\n"
+                if event.get("overall_status") in ("completed", "failed"):
+                    break
+        except asyncio.TimeoutError:
+            yield f"data: {json.dumps({'step': 'timeout', 'status': 'failed', 'overall_status': 'failed'})}\n\n"
+        finally:
+            manager.unsubscribe(job, queue)
+
+    return StreamingResponse(event_stream(), media_type="text/event-stream")
+
+
+@router.get("/jobs/{job_id}/assembly-tree", response_model=AssemblyTreeResponse)
+async def get_assembly_tree(job_id: str) -> AssemblyTreeResponse:
+    """Get the assembly tree for a job."""
+    manager = get_manager()
+    job = manager.get_job(job_id)
+    if job is None:
+        raise HTTPException(404, f"Job not found: {job_id}")
+    if job.assembly_tree is None:
+        raise HTTPException(409, "Assembly tree not available yet")
+
+    tree_dict = job.assembly_tree.to_dict()
+    root = _node_to_response(tree_dict)
+
+    groups = [
+        GroupInfoResponse(
+            name=g.name,
+            part_count=len(g.part_ids),
+            part_ids=g.part_ids,
+            part_names=g.part_names,
+        )
+        for g in job.groups.values()
+    ]
+
+    return AssemblyTreeResponse(root=root, groups=groups)
+
+
+@router.get("/jobs/{job_id}/parts/{part_id}/mesh")
+async def get_part_mesh(job_id: str, part_id: str) -> Response:
+    """Get the GLB mesh for a specific part."""
+    manager = get_manager()
+    job = manager.get_job(job_id)
+    if job is None:
+        raise HTTPException(404, f"Job not found: {job_id}")
+
+    mesh_data = job.part_meshes.get(part_id)
+    if mesh_data is None:
+        raise HTTPException(404, f"Part mesh not found: {part_id}")
+    if not mesh_data.glb:
+        raise HTTPException(409, "Mesh not exported yet")
+
+    return Response(content=mesh_data.glb, media_type="model/gltf-binary")
+
+
+@router.get("/jobs/{job_id}/groups/{group}/mesh")
+async def get_group_mesh(job_id: str, group: str) -> Response:
+    """Get the combined GLB mesh for a group (exterior or interior)."""
+    manager = get_manager()
+    job = manager.get_job(job_id)
+    if job is None:
+        raise HTTPException(404, f"Job not found: {job_id}")
+
+    if group not in ("exterior", "interior"):
+        raise HTTPException(400, f"Invalid group: {group}. Must be 'exterior' or 'interior'.")
+
+    mesh_data = job.group_meshes.get(group)
+    if mesh_data is None:
+        raise HTTPException(404, f"Group mesh not found: {group}")
+    if not mesh_data.glb:
+        raise HTTPException(409, "Group mesh not exported yet")
+
+    return Response(content=mesh_data.glb, media_type="model/gltf-binary")
+
+
+@router.post("/jobs/{job_id}/analyze/proximity", response_model=ProximityResponse)
+async def analyze_proximity(job_id: str) -> ProximityResponse:
+    """Run proximity/collision analysis on all parts."""
+    manager = get_manager()
+    job = manager.get_job(job_id)
+    if job is None:
+        raise HTTPException(404, f"Job not found: {job_id}")
+    if job.assembly_tree is None:
+        raise HTTPException(409, "Assembly tree not available yet")
+
+    parts = []
+    for leaf in job.assembly_tree.iter_leaves():
+        if leaf.shape is not None and not leaf.shape.IsNull():
+            parts.append({"id": leaf.id, "name": leaf.name, "shape": leaf.shape})
+
+    results = scan_proximity(
+        parts,
+        collision_threshold_mm=manager.settings.tolerance.gap_tolerance_mm,
+        near_threshold_mm=5.0,
+    )
+
+    pairs = [
+        ProximityPairResponse(
+            part_a_id=r.part_a_id,
+            part_a_name=r.part_a_name,
+            part_b_id=r.part_b_id,
+            part_b_name=r.part_b_name,
+            min_distance_mm=r.min_distance_mm,
+            point_a=r.point_a,
+            point_b=r.point_b,
+            status=r.status,
+        )
+        for r in results
+    ]
+
+    return ProximityResponse(
+        pairs=pairs,
+        collision_count=sum(1 for p in pairs if p.status == "collision"),
+        near_count=sum(1 for p in pairs if p.status == "near"),
+    )
+
+
+@router.post("/jobs/{job_id}/analyze/distance", response_model=DistanceMeasurementResponse)
+async def analyze_distance(job_id: str, request: DistanceRequest) -> DistanceMeasurementResponse:
+    """Measure distance between two specific parts."""
+    manager = get_manager()
+    job = manager.get_job(job_id)
+    if job is None:
+        raise HTTPException(404, f"Job not found: {job_id}")
+    if job.assembly_tree is None:
+        raise HTTPException(409, "Assembly tree not available yet")
+
+    node_a = job.assembly_tree.find_by_id(request.part_a_id)
+    node_b = job.assembly_tree.find_by_id(request.part_b_id)
+
+    if node_a is None or node_a.shape is None:
+        raise HTTPException(404, f"Part not found: {request.part_a_id}")
+    if node_b is None or node_b.shape is None:
+        raise HTTPException(404, f"Part not found: {request.part_b_id}")
+
+    result = measure_distance(node_a.shape, node_b.shape)
+
+    return DistanceMeasurementResponse(
+        distance_mm=result.distance_mm,
+        point_a=result.point_a,
+        point_b=result.point_b,
+    )
+
+
+@router.get("/compliance/rules")
+async def get_compliance_rules() -> list[ComplianceRuleResponse]:
+    """Get available compliance rules."""
+    rules_path = Path(__file__).parent.parent.parent / "config" / "kmvss_rules.yaml"
+    rules = load_rules(rules_path)
+    return [
+        ComplianceRuleResponse(
+            id=r.id, name=r.name, description=r.description,
+            type=r.type, severity=r.severity,
+        )
+        for r in rules
+    ]
+
+
+@router.post("/jobs/{job_id}/analyze/compliance", response_model=ComplianceResponse)
+async def analyze_compliance(job_id: str) -> ComplianceResponse:
+    """Run compliance checks on the assembly."""
+    manager = get_manager()
+    job = manager.get_job(job_id)
+    if job is None:
+        raise HTTPException(404, f"Job not found: {job_id}")
+    if job.assembly_tree is None:
+        raise HTTPException(409, "Assembly tree not available yet")
+
+    rules_path = Path(__file__).parent.parent.parent / "config" / "kmvss_rules.yaml"
+    rules = load_rules(rules_path)
+    results = check_compliance(rules, job.assembly_tree)
+
+    return ComplianceResponse(
+        results=[
+            ComplianceResultResponse(
+                rule_id=r.rule_id,
+                rule_name=r.rule_name,
+                passed=r.passed,
+                severity=r.severity,
+                measured_value=r.measured_value,
+                threshold_value=r.threshold_value,
+                unit=r.unit,
+                message=r.message,
+                affected_parts=r.affected_parts,
+            )
+            for r in results
+        ],
+        pass_count=sum(1 for r in results if r.passed),
+        fail_count=sum(1 for r in results if not r.passed),
+    )

src/api/samples.py

@@ -0,0 +1,75 @@
+"""Sample dataset discovery for the public example loader."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+_SUFFIX_EXTERIOR = "_exterior"
+_SUFFIX_INTERIOR = "_interior"
+_ALLOWED_EXT = (".step", ".stp")
+
+
+@dataclass(frozen=True)
+class SampleFile:
+    filename: str
+    path: Path
+    size: int
+
+
+@dataclass(frozen=True)
+class SamplePair:
+    id: str
+    name: str
+    exterior: SampleFile
+    interior: SampleFile
+
+
+def _pretty_name(stem: str) -> str:
+    return stem.replace("_", " ").replace("-", " ").strip().title()
+
+
+def _stem_for(path: Path, suffix: str) -> str | None:
+    stem = path.stem
+    return stem[: -len(suffix)] if stem.endswith(suffix) else None
+
+
+def discover_samples(samples_dir: Path) -> list[SamplePair]:
+    """Scan `samples_dir` for `{id}_exterior.(step|stp)` + `{id}_interior.(step|stp)` pairs."""
+    if not samples_dir.exists() or not samples_dir.is_dir():
+        return []
+
+    by_id_ext: dict[str, Path] = {}
+    by_id_int: dict[str, Path] = {}
+
+    for file in sorted(samples_dir.iterdir()):
+        if not file.is_file() or file.suffix.lower() not in _ALLOWED_EXT:
+            continue
+        if (sid := _stem_for(file, _SUFFIX_EXTERIOR)) is not None:
+            by_id_ext[sid] = file
+        elif (sid := _stem_for(file, _SUFFIX_INTERIOR)) is not None:
+            by_id_int[sid] = file
+
+    pairs: list[SamplePair] = []
+    for sid in sorted(set(by_id_ext) & set(by_id_int)):
+        ext_path = by_id_ext[sid]
+        int_path = by_id_int[sid]
+        pairs.append(
+            SamplePair(
+                id=sid,
+                name=_pretty_name(sid),
+                exterior=SampleFile(
+                    filename=ext_path.name, path=ext_path, size=ext_path.stat().st_size
+                ),
+                interior=SampleFile(
+                    filename=int_path.name, path=int_path, size=int_path.stat().st_size
+                ),
+            )
+        )
+    return pairs
+
+
+def get_sample(samples_dir: Path, sample_id: str) -> SamplePair | None:
+    for pair in discover_samples(samples_dir):
+        if pair.id == sample_id:
+            return pair
+    return None

src/api/schemas.py

@@ -0,0 +1,132 @@
+"""Pydantic request/response models for the CAD review API."""
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+# --- Request models ---
+
+class DistanceRequest(BaseModel):
+    """Request to measure distance between two parts."""
+    part_a_id: str
+    part_b_id: str
+
+
+# --- Response models ---
+
+class UploadResponse(BaseModel):
+    job_id: str
+    status: str
+    exterior_filename: str
+    interior_filename: str
+
+
+class SampleFileResponse(BaseModel):
+    filename: str
+    size: int
+
+
+class SamplePairResponse(BaseModel):
+    id: str
+    name: str
+    exterior: SampleFileResponse
+    interior: SampleFileResponse
+
+
+class StepProgress(BaseModel):
+    step: str
+    status: str  # pending, running, completed
+    progress_pct: int = 0
+
+
+class JobStatus(BaseModel):
+    job_id: str
+    status: str  # pending, running, completed, failed
+    steps: list[StepProgress] = []
+    error: str | None = None
+
+# Rebuild to resolve forward reference
+JobStatus.model_rebuild()
+
+
+class AssemblyNodeResponse(BaseModel):
+    id: str
+    name: str
+    is_assembly: bool
+    is_leaf: bool
+    classification: str = "unknown"
+    num_faces: int = 0
+    num_solids: int = 0
+    bounding_box: dict | None = None
+    children: list[AssemblyNodeResponse] = []
+
+AssemblyNodeResponse.model_rebuild()
+
+
+class GroupInfoResponse(BaseModel):
+    name: str  # "exterior" or "interior"
+    part_count: int
+    part_ids: list[str]
+    part_names: list[str]
+
+
+class AssemblyTreeResponse(BaseModel):
+    root: AssemblyNodeResponse
+    groups: list[GroupInfoResponse] = []
+
+
+class ProximityPairResponse(BaseModel):
+    part_a_id: str
+    part_a_name: str
+    part_b_id: str
+    part_b_name: str
+    min_distance_mm: float
+    point_a: list[float]
+    point_b: list[float]
+    status: str  # collision, near, ok
+
+
+class ProximityResponse(BaseModel):
+    pairs: list[ProximityPairResponse]
+    collision_count: int
+    near_count: int
+
+
+class DistanceMeasurementResponse(BaseModel):
+    distance_mm: float
+    point_a: list[float]
+    point_b: list[float]
+
+
+class ComplianceResultResponse(BaseModel):
+    rule_id: str
+    rule_name: str
+    passed: bool
+    severity: str
+    measured_value: float | None = None
+    threshold_value: float | None = None
+    unit: str = "mm"
+    message: str = ""
+    affected_parts: list[str] = []
+
+
+class ComplianceRuleResponse(BaseModel):
+    id: str
+    name: str
+    description: str
+    type: str
+    severity: str
+
+
+class ComplianceResponse(BaseModel):
+    results: list[ComplianceResultResponse]
+    pass_count: int
+    fail_count: int
+
+
+class SSEEvent(BaseModel):
+    step: str
+    status: str
+    progress_pct: int = 0
+    overall_status: str | None = None
+    error: str | None = None

src/comparison/continuity.py

@@ -0,0 +1,207 @@
+"""G0/G1/G2 surface continuity checking at boundary regions."""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+
+import numpy as np
+from OCP.BRep import BRep_Tool
+from OCP.BRepAdaptor import BRepAdaptor_Surface
+from OCP.GeomLProp import GeomLProp_SLProps
+from OCP.ShapeAnalysis import ShapeAnalysis_Surface
+from OCP.TopAbs import TopAbs_FACE
+from OCP.TopExp import TopExp_Explorer
+from OCP.TopoDS import TopoDS, TopoDS_Face, TopoDS_Shape
+from OCP.gp import gp_Pnt
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ContinuityResult:
+    """Results of continuity checks."""
+
+    g0_deviations: list[float] = field(default_factory=list)
+    g1_deviations_deg: list[float] = field(default_factory=list)
+    g2_deviations_pct: list[float] = field(default_factory=list)
+
+    g0_max: float = 0.0
+    g1_max_deg: float = 0.0
+    g2_max_pct: float = 0.0
+
+    g0_pass: bool = True
+    g1_pass: bool = True
+    g2_pass: bool = True
+
+    num_samples_checked: int = 0
+
+
+def _get_closest_face_and_uv(
+    faces: list[TopoDS_Face],
+    point: np.ndarray,
+) -> tuple[TopoDS_Face, float, float, float] | None:
+    """Find the closest face to a 3D point and return UV parameters."""
+    best_face = None
+    best_dist = float("inf")
+    best_u, best_v = 0.0, 0.0
+
+    gp_point = gp_Pnt(float(point[0]), float(point[1]), float(point[2]))
+
+    for face in faces:
+        try:
+            surface = BRep_Tool.Surface_s(face)
+            sas = ShapeAnalysis_Surface(surface)
+            uv = sas.ValueOfUV(gp_point, 1.0)
+            proj_pnt = sas.Value(uv.X(), uv.Y())
+            dist = gp_point.Distance(proj_pnt)
+
+            if dist < best_dist:
+                best_dist = dist
+                best_face = face
+                best_u = uv.X()
+                best_v = uv.Y()
+        except Exception:
+            continue
+
+    if best_face is None:
+        return None
+    return best_face, best_u, best_v, best_dist
+
+
+def _evaluate_surface_props(
+    face: TopoDS_Face,
+    u: float,
+    v: float,
+) -> tuple[np.ndarray | None, np.ndarray | None, float | None]:
+    """Evaluate surface normal and curvature at (u, v)."""
+    try:
+        surface = BRep_Tool.Surface_s(face)
+        props = GeomLProp_SLProps(surface, u, v, 2, 1e-6)
+
+        normal = None
+        if props.IsNormalDefined():
+            n = props.Normal()
+            normal = np.array([n.X(), n.Y(), n.Z()])
+
+        curvature = None
+        if props.IsCurvatureDefined():
+            curvature = props.MeanCurvature()
+
+        point = props.Value()
+        position = np.array([point.X(), point.Y(), point.Z()])
+
+        return position, normal, curvature
+    except Exception:
+        return None, None, None
+
+
+def _collect_faces(shape: TopoDS_Shape) -> list[TopoDS_Face]:
+    """Collect all faces from a shape."""
+    faces = []
+    exp = TopExp_Explorer(shape, TopAbs_FACE)
+    while exp.More():
+        faces.append(TopoDS.Face_s(exp.Current()))
+        exp.Next()
+    return faces
+
+
+def check_continuity(
+    shape_a: TopoDS_Shape,
+    shape_b: TopoDS_Shape,
+    sample_points: np.ndarray,
+    g0_tolerance_mm: float = 0.05,
+    g1_tolerance_deg: float = 0.3,
+    g2_tolerance_pct: float = 3.0,
+    max_check_points: int = 1000,
+) -> ContinuityResult:
+    """Check G0/G1/G2 continuity between two shapes at sample points.
+
+    Args:
+        shape_a: First shape (reference).
+        shape_b: Second shape (comparison).
+        sample_points: Points at which to evaluate continuity.
+        g0_tolerance_mm: G0 positional tolerance.
+        g1_tolerance_deg: G1 tangent angle tolerance.
+        g2_tolerance_pct: G2 curvature relative tolerance.
+        max_check_points: Limit on number of points to check.
+
+    Returns:
+        ContinuityResult with pass/fail per grade.
+    """
+    faces_a = _collect_faces(shape_a)
+    faces_b = _collect_faces(shape_b)
+
+    if not faces_a or not faces_b:
+        logger.warning("One or both shapes have no faces -- skipping continuity check")
+        return ContinuityResult()
+
+    # Subsample if needed
+    check_points = sample_points
+    if len(sample_points) > max_check_points:
+        rng = np.random.default_rng(42)
+        indices = rng.choice(len(sample_points), max_check_points, replace=False)
+        check_points = sample_points[indices]
+
+    result = ContinuityResult()
+
+    for point in check_points:
+        match_a = _get_closest_face_and_uv(faces_a, point)
+        match_b = _get_closest_face_and_uv(faces_b, point)
+
+        if match_a is None or match_b is None:
+            continue
+
+        face_a, u_a, v_a, dist_a = match_a
+        face_b, u_b, v_b, dist_b = match_b
+
+        # Skip if projection is too far (point not on either surface)
+        if dist_a > g0_tolerance_mm * 10 or dist_b > g0_tolerance_mm * 10:
+            continue
+
+        pos_a, norm_a, curv_a = _evaluate_surface_props(face_a, u_a, v_a)
+        pos_b, norm_b, curv_b = _evaluate_surface_props(face_b, u_b, v_b)
+
+        if pos_a is None or pos_b is None:
+            continue
+
+        result.num_samples_checked += 1
+
+        # G0: positional
+        g0_dev = float(np.linalg.norm(pos_a - pos_b))
+        result.g0_deviations.append(g0_dev)
+
+        # G1: tangent (normal angle)
+        if norm_a is not None and norm_b is not None:
+            cos_angle = np.clip(np.dot(norm_a, norm_b), -1.0, 1.0)
+            angle_deg = float(np.degrees(np.arccos(abs(cos_angle))))
+            result.g1_deviations_deg.append(angle_deg)
+
+        # G2: curvature
+        if curv_a is not None and curv_b is not None:
+            denom = max(abs(curv_a), abs(curv_b), 1e-12)
+            g2_pct = abs(curv_a - curv_b) / denom * 100.0
+            result.g2_deviations_pct.append(float(g2_pct))
+
+    # Compute max and pass/fail
+    if result.g0_deviations:
+        result.g0_max = max(result.g0_deviations)
+        result.g0_pass = result.g0_max <= g0_tolerance_mm
+
+    if result.g1_deviations_deg:
+        result.g1_max_deg = max(result.g1_deviations_deg)
+        result.g1_pass = result.g1_max_deg <= g1_tolerance_deg
+
+    if result.g2_deviations_pct:
+        result.g2_max_pct = max(result.g2_deviations_pct)
+        result.g2_pass = result.g2_max_pct <= g2_tolerance_pct
+
+    logger.info(
+        "Continuity check (%d points): G0 max=%.4f mm (%s), "
+        "G1 max=%.2f deg (%s), G2 max=%.1f%% (%s)",
+        result.num_samples_checked,
+        result.g0_max, "PASS" if result.g0_pass else "FAIL",
+        result.g1_max_deg, "PASS" if result.g1_pass else "FAIL",
+        result.g2_max_pct, "PASS" if result.g2_pass else "FAIL",
+    )
+    return result

src/comparison/distance.py

@@ -0,0 +1,101 @@
+"""Bidirectional distance computation between point clouds."""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+
+import numpy as np
+import open3d as o3d
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class DistanceResult:
+    """Results of bidirectional distance computation."""
+
+    # A -> B distances
+    a_to_b_distances: np.ndarray  # per-point distances
+    a_to_b_max: float  # Hausdorff (one direction)
+    a_to_b_rms: float
+    a_to_b_mean: float
+
+    # B -> A distances
+    b_to_a_distances: np.ndarray
+    b_to_a_max: float
+    b_to_a_rms: float
+    b_to_a_mean: float
+
+    # Combined metrics
+    hausdorff: float  # max(a_to_b_max, b_to_a_max)
+    rms: float  # RMS of combined
+    mean: float
+    percentile_95: float
+    percentile_99: float
+
+
+def _one_directional_distances(
+    source: np.ndarray,
+    target: np.ndarray,
+) -> np.ndarray:
+    """Compute nearest-neighbor distances from source to target.
+
+    Uses Open3D KDTree for efficient lookup.
+    """
+    target_pcd = o3d.geometry.PointCloud()
+    target_pcd.points = o3d.utility.Vector3dVector(target)
+    kdtree = o3d.geometry.KDTreeFlann(target_pcd)
+
+    distances = np.empty(len(source), dtype=np.float64)
+    for i, point in enumerate(source):
+        _, _, dist_sq = kdtree.search_knn_vector_3d(point, 1)
+        distances[i] = np.sqrt(dist_sq[0])
+
+    return distances
+
+
+def compute_bidirectional_distances(
+    points_a: np.ndarray,
+    points_b: np.ndarray,
+) -> DistanceResult:
+    """Compute bidirectional nearest-neighbor distances.
+
+    Args:
+        points_a: Point cloud A [N, 3].
+        points_b: Point cloud B [M, 3].
+
+    Returns:
+        DistanceResult with per-point distances and aggregate metrics.
+    """
+    logger.info("Computing A->B distances (%d -> %d points)", len(points_a), len(points_b))
+    a_to_b = _one_directional_distances(points_a, points_b)
+
+    logger.info("Computing B->A distances (%d -> %d points)", len(points_b), len(points_a))
+    b_to_a = _one_directional_distances(points_b, points_a)
+
+    combined = np.concatenate([a_to_b, b_to_a])
+
+    result = DistanceResult(
+        a_to_b_distances=a_to_b,
+        a_to_b_max=float(np.max(a_to_b)),
+        a_to_b_rms=float(np.sqrt(np.mean(a_to_b**2))),
+        a_to_b_mean=float(np.mean(a_to_b)),
+        b_to_a_distances=b_to_a,
+        b_to_a_max=float(np.max(b_to_a)),
+        b_to_a_rms=float(np.sqrt(np.mean(b_to_a**2))),
+        b_to_a_mean=float(np.mean(b_to_a)),
+        hausdorff=float(max(np.max(a_to_b), np.max(b_to_a))),
+        rms=float(np.sqrt(np.mean(combined**2))),
+        mean=float(np.mean(combined)),
+        percentile_95=float(np.percentile(combined, 95)),
+        percentile_99=float(np.percentile(combined, 99)),
+    )
+
+    logger.info(
+        "Distance results: Hausdorff=%.6f mm, RMS=%.6f mm, Mean=%.6f mm, "
+        "P95=%.6f mm, P99=%.6f mm",
+        result.hausdorff, result.rms, result.mean,
+        result.percentile_95, result.percentile_99,
+    )
+    return result

src/comparison/gap_overlap.py

@@ -0,0 +1,156 @@
+"""Gap and overlap detection on boundary edges."""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+
+import numpy as np
+import open3d as o3d
+from OCP.BRep import BRep_Tool
+from OCP.BRepAdaptor import BRepAdaptor_Curve
+from OCP.GCPnts import GCPnts_UniformAbscissa
+from OCP.TopAbs import TopAbs_EDGE, TopAbs_FACE
+from OCP.TopExp import TopExp_Explorer
+from OCP.TopoDS import TopoDS, TopoDS_Shape
+
+logger = logging.getLogger(__name__)
+
+NUM_EDGE_SAMPLES = 50
+
+
+@dataclass
+class GapOverlapResult:
+    """Gap and overlap detection results."""
+
+    gap_distances: list[float] = field(default_factory=list)
+    overlap_distances: list[float] = field(default_factory=list)
+    gap_locations: list[np.ndarray] = field(default_factory=list)
+    overlap_locations: list[np.ndarray] = field(default_factory=list)
+
+    @property
+    def gap_count(self) -> int:
+        return len(self.gap_distances)
+
+    @property
+    def overlap_count(self) -> int:
+        return len(self.overlap_distances)
+
+    @property
+    def max_gap(self) -> float:
+        return max(self.gap_distances) if self.gap_distances else 0.0
+
+    @property
+    def max_overlap(self) -> float:
+        return max(self.overlap_distances) if self.overlap_distances else 0.0
+
+
+def _sample_boundary_edges(shape: TopoDS_Shape) -> np.ndarray:
+    """Sample points along boundary edges of a shape.
+
+    Boundary edges are edges that belong to only one face.
+    """
+    # Collect edge-to-face counts
+    edge_face_count: dict[int, int] = {}
+    edge_map: dict[int, object] = {}
+
+    face_exp = TopExp_Explorer(shape, TopAbs_FACE)
+    while face_exp.More():
+        face = TopoDS.Face_s(face_exp.Current())
+        edge_exp = TopExp_Explorer(face, TopAbs_EDGE)
+        while edge_exp.More():
+            edge = TopoDS.Edge_s(edge_exp.Current())
+            h = hash(edge)
+            edge_face_count[h] = edge_face_count.get(h, 0) + 1
+            edge_map[h] = edge
+            edge_exp.Next()
+        face_exp.Next()
+
+    # Boundary edges: shared by only one face
+    boundary_edges = [edge_map[h] for h, count in edge_face_count.items() if count == 1]
+    logger.info("Found %d boundary edges out of %d total", len(boundary_edges), len(edge_map))
+
+    if not boundary_edges:
+        return np.empty((0, 3), dtype=np.float64)
+
+    points = []
+    for edge in boundary_edges:
+        try:
+            curve = BRepAdaptor_Curve(edge)
+            u_first = curve.FirstParameter()
+            u_last = curve.LastParameter()
+
+            distributor = GCPnts_UniformAbscissa(curve, NUM_EDGE_SAMPLES, u_first, u_last)
+            if not distributor.IsDone():
+                continue
+
+            for i in range(1, distributor.NbPoints() + 1):
+                param = distributor.Parameter(i)
+                pnt = curve.Value(param)
+                points.append([pnt.X(), pnt.Y(), pnt.Z()])
+        except Exception:
+            continue
+
+    if not points:
+        return np.empty((0, 3), dtype=np.float64)
+
+    return np.array(points, dtype=np.float64)
+
+
+def detect_gaps_and_overlaps(
+    shape_a: TopoDS_Shape,
+    shape_b: TopoDS_Shape,
+    vertices_b: np.ndarray,
+    gap_tolerance_mm: float = 0.05,
+    overlap_tolerance_mm: float = 0.05,
+) -> GapOverlapResult:
+    """Detect gaps and overlaps by projecting boundary edge points.
+
+    Samples boundary edges of shape_a and measures distance to
+    the closest point on shape_b's mesh surface.
+
+    Args:
+        shape_a: First shape (will sample boundary edges from).
+        shape_b: Second shape (reference).
+        vertices_b: Tessellated vertices of shape_b.
+        gap_tolerance_mm: Gap detection threshold.
+        overlap_tolerance_mm: Overlap detection threshold.
+
+    Returns:
+        GapOverlapResult with detected gaps and overlaps.
+    """
+    boundary_points = _sample_boundary_edges(shape_a)
+
+    if len(boundary_points) == 0:
+        logger.info("No boundary edges found -- skipping gap/overlap detection")
+        return GapOverlapResult()
+
+    if len(vertices_b) == 0:
+        logger.warning("Target mesh has no vertices -- skipping gap/overlap detection")
+        return GapOverlapResult()
+
+    # Build KDTree on target mesh
+    target_pcd = o3d.geometry.PointCloud()
+    target_pcd.points = o3d.utility.Vector3dVector(vertices_b)
+    kdtree = o3d.geometry.KDTreeFlann(target_pcd)
+
+    result = GapOverlapResult()
+
+    for point in boundary_points:
+        _, idx, dist_sq = kdtree.search_knn_vector_3d(point, 1)
+        dist = np.sqrt(dist_sq[0])
+
+        if dist > gap_tolerance_mm:
+            result.gap_distances.append(float(dist))
+            result.gap_locations.append(point)
+        elif dist < overlap_tolerance_mm * 0.1:
+            # Very close = potential overlap (surface interpenetration)
+            result.overlap_distances.append(float(dist))
+            result.overlap_locations.append(point)
+
+    logger.info(
+        "Gap/overlap detection: %d gaps (max=%.4f mm), %d overlaps (max=%.4f mm)",
+        result.gap_count, result.max_gap,
+        result.overlap_count, result.max_overlap,
+    )
+    return result

src/compliance/kmvss_checker.py

@@ -0,0 +1,189 @@
+"""KMVSS compliance checker - rule engine for geometry validation."""
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass, field
+
+import numpy as np
+from OCP.Bnd import Bnd_Box
+from OCP.BRepBndLib import BRepBndLib
+from OCP.TopoDS import TopoDS_Shape
+
+from src.compliance.rule_loader import ComplianceRule
+from src.geometry.measurement import measure_distance
+from src.loader.assembly_tree import AssemblyNode
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ComplianceResult:
+    """Result of a single compliance check."""
+    rule_id: str
+    rule_name: str
+    passed: bool
+    severity: str
+    measured_value: float | None = None
+    threshold_value: float | None = None
+    unit: str = "mm"
+    message: str = ""
+    affected_parts: list[str] = field(default_factory=list)
+
+
+def _find_parts_by_pattern(root: AssemblyNode, pattern: str) -> list[AssemblyNode]:
+    """Find leaf parts whose name matches a regex pattern."""
+    regex = re.compile(pattern, re.IGNORECASE)
+    return [leaf for leaf in root.iter_leaves()
+            if leaf.shape is not None and regex.search(leaf.name)]
+
+
+def _get_bbox_dimension(shape: TopoDS_Shape, axis: str) -> float:
+    """Get a bounding box dimension along an axis."""
+    bbox = Bnd_Box()
+    BRepBndLib.Add_s(shape, bbox)
+    xmin, ymin, zmin, xmax, ymax, zmax = bbox.Get()
+
+    axis_map = {
+        "x": xmax - xmin,
+        "y": ymax - ymin,
+        "z": zmax - zmin,
+    }
+    return axis_map.get(axis.lower(), 0.0)
+
+
+def _check_bounding_box(
+    rule: ComplianceRule,
+    root: AssemblyNode,
+) -> ComplianceResult:
+    """Check bounding box dimension constraint."""
+    parts = _find_parts_by_pattern(root, rule.params.target_name_pattern)
+
+    if not parts:
+        return ComplianceResult(
+            rule_id=rule.id,
+            rule_name=rule.name,
+            passed=True,
+            severity=rule.severity,
+            message=f"No parts matching '{rule.params.target_name_pattern}' found (skipped)",
+        )
+
+    max_dim = 0.0
+    for part in parts:
+        dim = _get_bbox_dimension(part.shape, rule.params.axis)
+        max_dim = max(max_dim, dim)
+
+    passed = max_dim <= rule.params.max_mm
+
+    return ComplianceResult(
+        rule_id=rule.id,
+        rule_name=rule.name,
+        passed=passed,
+        severity=rule.severity,
+        measured_value=max_dim,
+        threshold_value=rule.params.max_mm,
+        message=f"{rule.params.axis.upper()}-axis dimension: {max_dim:.1f} mm (max: {rule.params.max_mm:.1f} mm)",
+        affected_parts=[p.name for p in parts],
+    )
+
+
+def _check_clearance(
+    rule: ComplianceRule,
+    root: AssemblyNode,
+) -> ComplianceResult:
+    """Check minimum clearance between two part groups."""
+    parts_a = _find_parts_by_pattern(root, rule.params.part_a_pattern)
+    parts_b = _find_parts_by_pattern(root, rule.params.part_b_pattern)
+
+    if not parts_a or not parts_b:
+        missing = []
+        if not parts_a:
+            missing.append(rule.params.part_a_pattern)
+        if not parts_b:
+            missing.append(rule.params.part_b_pattern)
+        return ComplianceResult(
+            rule_id=rule.id,
+            rule_name=rule.name,
+            passed=True,
+            severity=rule.severity,
+            message=f"Parts not found for pattern(s): {', '.join(missing)} (skipped)",
+        )
+
+    min_clearance = float("inf")
+    affected = []
+
+    for pa in parts_a:
+        for pb in parts_b:
+            try:
+                result = measure_distance(pa.shape, pb.shape)
+                if result.distance_mm < min_clearance:
+                    min_clearance = result.distance_mm
+                    affected = [pa.name, pb.name]
+            except Exception as e:
+                logger.warning("Clearance check failed for %s vs %s: %s", pa.name, pb.name, e)
+
+    if min_clearance == float("inf"):
+        return ComplianceResult(
+            rule_id=rule.id,
+            rule_name=rule.name,
+            passed=True,
+            severity=rule.severity,
+            message="Could not measure clearance (skipped)",
+        )
+
+    passed = min_clearance >= rule.params.min_clearance_mm
+
+    return ComplianceResult(
+        rule_id=rule.id,
+        rule_name=rule.name,
+        passed=passed,
+        severity=rule.severity,
+        measured_value=min_clearance,
+        threshold_value=rule.params.min_clearance_mm,
+        unit="mm",
+        message=f"Min clearance: {min_clearance:.2f} mm (required: >= {rule.params.min_clearance_mm:.1f} mm)",
+        affected_parts=affected,
+    )
+
+
+def check_compliance(
+    rules: list[ComplianceRule],
+    root: AssemblyNode,
+) -> list[ComplianceResult]:
+    """Run all compliance rules against the assembly.
+
+    Args:
+        rules: List of compliance rules to check.
+        root: Assembly tree root node.
+
+    Returns:
+        List of ComplianceResult for each rule.
+    """
+    results = []
+
+    for rule in rules:
+        logger.info("Checking rule: %s (%s)", rule.name, rule.type)
+
+        if rule.type == "bounding_box":
+            result = _check_bounding_box(rule, root)
+        elif rule.type == "clearance":
+            result = _check_clearance(rule, root)
+        else:
+            result = ComplianceResult(
+                rule_id=rule.id,
+                rule_name=rule.name,
+                passed=True,
+                severity=rule.severity,
+                message=f"Unknown rule type: {rule.type} (skipped)",
+            )
+
+        results.append(result)
+        status = "PASS" if result.passed else "FAIL"
+        logger.info("  -> %s: %s", status, result.message)
+
+    pass_count = sum(1 for r in results if r.passed)
+    fail_count = len(results) - pass_count
+    logger.info("Compliance: %d passed, %d failed out of %d rules",
+                pass_count, fail_count, len(results))
+
+    return results

src/compliance/rule_loader.py

@@ -0,0 +1,66 @@
+"""Load compliance rules from YAML configuration."""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import yaml
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class RuleParam:
+    """Parameters for a compliance rule."""
+    target_name_pattern: str = ""
+    part_a_pattern: str = ""
+    part_b_pattern: str = ""
+    axis: str = ""
+    max_mm: float = 0.0
+    min_clearance_mm: float = 0.0
+
+
+@dataclass
+class ComplianceRule:
+    """A single compliance rule definition."""
+    id: str
+    name: str
+    description: str
+    type: str  # bounding_box, clearance, part_height
+    params: RuleParam
+    severity: str = "error"  # error, warning, info
+
+
+def load_rules(yaml_path: str | Path) -> list[ComplianceRule]:
+    """Load compliance rules from a YAML file.
+
+    Args:
+        yaml_path: Path to the YAML rules file.
+
+    Returns:
+        List of ComplianceRule objects.
+    """
+    yaml_path = Path(yaml_path)
+    if not yaml_path.exists():
+        logger.warning("Rules file not found: %s", yaml_path)
+        return []
+
+    with open(yaml_path) as f:
+        data = yaml.safe_load(f)
+
+    rules = []
+    for rule_data in data.get("rules", []):
+        params = RuleParam(**rule_data.get("params", {}))
+        rule = ComplianceRule(
+            id=rule_data["id"],
+            name=rule_data["name"],
+            description=rule_data.get("description", ""),
+            type=rule_data["type"],
+            params=params,
+            severity=rule_data.get("severity", "error"),
+        )
+        rules.append(rule)
+
+    logger.info("Loaded %d compliance rules from %s", len(rules), yaml_path)
+    return rules

src/geometry/face_extractor.py

@@ -0,0 +1,107 @@
+"""B-Rep face extraction: solid shells vs free surfaces."""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+
+from OCP.BRep import BRep_Tool
+from OCP.TopAbs import TopAbs_FACE, TopAbs_SHELL, TopAbs_SOLID
+from OCP.TopExp import TopExp_Explorer
+from OCP.TopoDS import TopoDS, TopoDS_Face, TopoDS_Shape
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class FaceExtractionResult:
+    """Result of face extraction from a shape."""
+
+    faces: list[TopoDS_Face]
+    is_solid: bool
+    num_solids: int
+    num_shells: int
+    num_faces: int
+
+
+def _count_topology(shape: TopoDS_Shape, topo_type) -> int:
+    """Count topology entities of a given type."""
+    count = 0
+    exp = TopExp_Explorer(shape, topo_type)
+    while exp.More():
+        count += 1
+        exp.Next()
+    return count
+
+
+def _has_surface_geometry(face: TopoDS_Face) -> bool:
+    """Check that a face has valid underlying surface geometry."""
+    try:
+        surface = BRep_Tool.Surface_s(face)
+        return surface is not None
+    except Exception:
+        return False
+
+
+def extract_faces(shape: TopoDS_Shape) -> FaceExtractionResult:
+    """Extract faces from a B-Rep shape.
+
+    For solids (CATIA): extracts outer shell faces.
+    For free surfaces (Alias): collects all faces directly.
+
+    Args:
+        shape: The TopoDS_Shape to extract faces from.
+
+    Returns:
+        FaceExtractionResult with the list of faces and metadata.
+    """
+    num_solids = _count_topology(shape, TopAbs_SOLID)
+    num_shells = _count_topology(shape, TopAbs_SHELL)
+    is_solid = num_solids > 0
+
+    faces: list[TopoDS_Face] = []
+    seen_hashes: set[int] = set()
+
+    if is_solid:
+        logger.info(
+            "Solid geometry detected (%d solid(s), %d shell(s)). "
+            "Extracting outer shell faces.",
+            num_solids, num_shells,
+        )
+        solid_exp = TopExp_Explorer(shape, TopAbs_SOLID)
+        while solid_exp.More():
+            shell_exp = TopExp_Explorer(solid_exp.Current(), TopAbs_FACE)
+            while shell_exp.More():
+                face = TopoDS.Face_s(shell_exp.Current())
+                h = hash(face)
+                if h not in seen_hashes and _has_surface_geometry(face):
+                    seen_hashes.add(h)
+                    faces.append(face)
+                shell_exp.Next()
+            solid_exp.Next()
+    else:
+        logger.info(
+            "Free-surface geometry detected (%d shell(s)). "
+            "Collecting all faces.",
+            num_shells,
+        )
+        face_exp = TopExp_Explorer(shape, TopAbs_FACE)
+        while face_exp.More():
+            face = TopoDS.Face_s(face_exp.Current())
+            h = hash(face)
+            if h not in seen_hashes and _has_surface_geometry(face):
+                seen_hashes.add(h)
+                faces.append(face)
+            face_exp.Next()
+
+    num_faces = len(faces)
+    logger.info("Extracted %d unique faces (duplicates removed: %s).",
+                num_faces, "solid" if is_solid else "surface")
+
+    return FaceExtractionResult(
+        faces=faces,
+        is_solid=is_solid,
+        num_solids=num_solids,
+        num_shells=num_shells,
+        num_faces=num_faces,
+    )

src/geometry/measurement.py

@@ -0,0 +1,130 @@
+"""Geometry measurement using BRepExtrema."""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+
+import numpy as np
+from OCP.BRepExtrema import BRepExtrema_DistShapeShape
+from OCP.TopoDS import TopoDS_Shape
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class DistanceMeasurement:
+    """Result of distance measurement between two shapes."""
+    distance_mm: float
+    point_a: list[float]  # Closest point on shape A
+    point_b: list[float]  # Closest point on shape B
+    num_solutions: int
+
+
+@dataclass
+class ProximityPair:
+    """A pair of parts with their proximity information."""
+    part_a_id: str
+    part_a_name: str
+    part_b_id: str
+    part_b_name: str
+    min_distance_mm: float
+    point_a: list[float]
+    point_b: list[float]
+    status: str  # "collision", "near", "ok"
+
+
+def measure_distance(
+    shape_a: TopoDS_Shape,
+    shape_b: TopoDS_Shape,
+) -> DistanceMeasurement:
+    """Measure minimum distance between two shapes using BRepExtrema.
+
+    Args:
+        shape_a: First shape.
+        shape_b: Second shape.
+
+    Returns:
+        DistanceMeasurement with distance and closest points.
+    """
+    dist_calc = BRepExtrema_DistShapeShape(shape_a, shape_b)
+    dist_calc.Perform()
+
+    if not dist_calc.IsDone():
+        raise RuntimeError("Distance computation failed")
+
+    distance = dist_calc.Value()
+    n_solutions = dist_calc.NbSolution()
+
+    if n_solutions > 0:
+        pt_a = dist_calc.PointOnShape1(1)
+        pt_b = dist_calc.PointOnShape2(1)
+        point_a = [pt_a.X(), pt_a.Y(), pt_a.Z()]
+        point_b = [pt_b.X(), pt_b.Y(), pt_b.Z()]
+    else:
+        point_a = [0, 0, 0]
+        point_b = [0, 0, 0]
+
+    logger.info("Distance: %.4f mm (%d solutions)", distance, n_solutions)
+
+    return DistanceMeasurement(
+        distance_mm=distance,
+        point_a=point_a,
+        point_b=point_b,
+        num_solutions=n_solutions,
+    )
+
+
+def scan_proximity(
+    parts: list[dict],
+    collision_threshold_mm: float = 0.1,
+    near_threshold_mm: float = 5.0,
+    max_pairs: int = 100,
+) -> list[ProximityPair]:
+    """Scan all part pairs for proximity/collision.
+
+    Args:
+        parts: List of dicts with 'id', 'name', 'shape' keys.
+        collision_threshold_mm: Distance below which parts are considered colliding.
+        near_threshold_mm: Distance below which parts are considered near.
+        max_pairs: Maximum number of pairs to return.
+
+    Returns:
+        List of ProximityPair sorted by distance.
+    """
+    results = []
+    n = len(parts)
+
+    logger.info("Scanning proximity for %d parts (%d pairs)", n, n * (n - 1) // 2)
+
+    for i in range(n):
+        for j in range(i + 1, n):
+            try:
+                measurement = measure_distance(parts[i]["shape"], parts[j]["shape"])
+
+                if measurement.distance_mm <= near_threshold_mm:
+                    if measurement.distance_mm <= collision_threshold_mm:
+                        status = "collision"
+                    else:
+                        status = "near"
+
+                    results.append(ProximityPair(
+                        part_a_id=parts[i]["id"],
+                        part_a_name=parts[i]["name"],
+                        part_b_id=parts[j]["id"],
+                        part_b_name=parts[j]["name"],
+                        min_distance_mm=measurement.distance_mm,
+                        point_a=measurement.point_a,
+                        point_b=measurement.point_b,
+                        status=status,
+                    ))
+            except Exception as e:
+                logger.warning("Failed to measure distance between %s and %s: %s",
+                             parts[i]["name"], parts[j]["name"], e)
+
+    results.sort(key=lambda p: p.min_distance_mm)
+
+    collision_count = sum(1 for r in results if r.status == "collision")
+    near_count = sum(1 for r in results if r.status == "near")
+    logger.info("Proximity scan: %d collisions, %d near pairs", collision_count, near_count)
+
+    return results[:max_pairs]

src/geometry/splitter.py

@@ -0,0 +1,84 @@
+"""Exterior/interior part classification for assembly review.
+
+Classifies parts based on name-pattern matching against configurable
+exterior patterns. Everything that does not match is classified as interior.
+"""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+
+from config.settings import SplitterSettings
+from src.loader.assembly_tree import AssemblyNode
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ClassificationResult:
+    """Result of exterior/interior classification for a part."""
+    node_id: str
+    name: str
+    classification: str  # "exterior" or "interior"
+    score: int  # 1 = matched exterior pattern, 0 = no match
+    reasons: list[str]
+
+
+def classify_parts(
+    root: AssemblyNode,
+    settings: SplitterSettings | None = None,
+) -> list[ClassificationResult]:
+    """Classify all leaf parts as exterior or interior.
+
+    Uses simple case-insensitive substring matching against the configured
+    exterior_patterns. Any part whose name matches at least one pattern
+    is classified as "exterior"; everything else is "interior".
+
+    Args:
+        root: Assembly tree root node.
+        settings: Splitter settings with exterior_patterns.
+
+    Returns:
+        List of ClassificationResult for each leaf part.
+    """
+    if settings is None:
+        settings = SplitterSettings()
+
+    patterns = [p.lower() for p in settings.exterior_patterns]
+
+    results: list[ClassificationResult] = []
+
+    for leaf in root.iter_leaves():
+        if leaf.shape is None or leaf.shape.IsNull():
+            continue
+
+        name_lower = leaf.name.lower()
+        matched = [p for p in patterns if p in name_lower]
+
+        if matched:
+            classification = "exterior"
+            score = 1
+            reasons = [f"name matches exterior pattern(s): {', '.join(matched)}"]
+        else:
+            classification = "interior"
+            score = 0
+            reasons = ["no exterior pattern matched"]
+
+        leaf.classification = classification
+
+        result = ClassificationResult(
+            node_id=leaf.id,
+            name=leaf.name,
+            classification=classification,
+            score=score,
+            reasons=reasons,
+        )
+        results.append(result)
+
+        logger.debug("Part '%s': -> %s (%s)", leaf.name, classification, "; ".join(reasons))
+
+    ext_count = sum(1 for r in results if r.classification == "exterior")
+    int_count = sum(1 for r in results if r.classification == "interior")
+    logger.info("Classification: %d exterior, %d interior", ext_count, int_count)
+
+    return results

src/geometry/tessellator.py

@@ -0,0 +1,133 @@
+"""Mesh tessellation and surface point sampling."""
+from __future__ import annotations
+
+import logging
+
+import numpy as np
+from OCP.BRep import BRep_Tool
+from OCP.BRepMesh import BRepMesh_IncrementalMesh
+from OCP.TopAbs import TopAbs_FACE
+from OCP.TopExp import TopExp_Explorer
+from OCP.TopLoc import TopLoc_Location
+from OCP.TopoDS import TopoDS, TopoDS_Face, TopoDS_Shape
+
+from config.settings import SamplingSettings, TessellationSettings
+
+logger = logging.getLogger(__name__)
+
+
+def tessellate_shape(
+    shape: TopoDS_Shape,
+    settings: TessellationSettings,
+    target_tolerance_mm: float = 0.05,
+) -> None:
+    """Tessellate a shape in-place."""
+    deflection = max(
+        settings.min_deflection_mm,
+        min(settings.max_deflection_mm, target_tolerance_mm * settings.deflection_factor),
+    )
+    logger.info("Tessellating with deflection=%.4f mm", deflection)
+
+    mesh = BRepMesh_IncrementalMesh(shape, deflection, False, settings.angular_deflection_rad, True)
+    mesh.Perform()
+    if not mesh.IsDone():
+        raise RuntimeError("Tessellation failed")
+
+
+def _extract_face_mesh(face: TopoDS_Face) -> tuple[np.ndarray, np.ndarray] | None:
+    """Extract vertices and triangle indices from a tessellated face."""
+    loc = TopLoc_Location()
+    triangulation = BRep_Tool.Triangulation_s(face, loc)
+    if triangulation is None:
+        return None
+
+    trsf = loc.Transformation()
+    n_nodes = triangulation.NbNodes()
+    n_tris = triangulation.NbTriangles()
+
+    if n_nodes == 0 or n_tris == 0:
+        return None
+
+    vertices = np.empty((n_nodes, 3), dtype=np.float64)
+    for i in range(1, n_nodes + 1):
+        node = triangulation.Node(i)
+        node_transformed = node.Transformed(trsf)
+        vertices[i - 1] = [node_transformed.X(), node_transformed.Y(), node_transformed.Z()]
+
+    triangles = np.empty((n_tris, 3), dtype=np.int32)
+    for i in range(1, n_tris + 1):
+        tri = triangulation.Triangle(i)
+        n1, n2, n3 = tri.Get()
+        triangles[i - 1] = [n1 - 1, n2 - 1, n3 - 1]
+
+    return vertices, triangles
+
+
+def tessellate_faces(
+    shape: TopoDS_Shape,
+    settings: TessellationSettings,
+    target_tolerance_mm: float = 0.05,
+) -> tuple[np.ndarray, np.ndarray]:
+    """Tessellate all faces and return combined mesh."""
+    tessellate_shape(shape, settings, target_tolerance_mm)
+
+    all_vertices: list[np.ndarray] = []
+    all_triangles: list[np.ndarray] = []
+    vertex_offset = 0
+
+    face_exp = TopExp_Explorer(shape, TopAbs_FACE)
+    while face_exp.More():
+        face = TopoDS.Face_s(face_exp.Current())
+        result = _extract_face_mesh(face)
+        if result is not None:
+            verts, tris = result
+            all_vertices.append(verts)
+            all_triangles.append(tris + vertex_offset)
+            vertex_offset += len(verts)
+        face_exp.Next()
+
+    if not all_vertices:
+        raise RuntimeError("No mesh data extracted from tessellation")
+
+    vertices = np.concatenate(all_vertices, axis=0)
+    triangles = np.concatenate(all_triangles, axis=0)
+    logger.info("Tessellated mesh: %d vertices, %d triangles", len(vertices), len(triangles))
+    return vertices, triangles
+
+
+def sample_points_on_mesh(
+    vertices: np.ndarray,
+    triangles: np.ndarray,
+    settings: SamplingSettings,
+) -> np.ndarray:
+    """Area-weighted uniform sampling on a triangle mesh."""
+    rng = np.random.default_rng(settings.seed)
+
+    v0 = vertices[triangles[:, 0]]
+    v1 = vertices[triangles[:, 1]]
+    v2 = vertices[triangles[:, 2]]
+
+    cross = np.cross(v1 - v0, v2 - v0)
+    areas = 0.5 * np.linalg.norm(cross, axis=1)
+    total_area = areas.sum()
+    if total_area < 1e-15:
+        raise RuntimeError("Degenerate mesh: total area is zero")
+
+    probabilities = areas / total_area
+    tri_indices = rng.choice(len(triangles), size=settings.num_samples, p=probabilities)
+
+    r1 = rng.random(settings.num_samples)
+    r2 = rng.random(settings.num_samples)
+    sqrt_r1 = np.sqrt(r1)
+
+    u = 1.0 - sqrt_r1
+    v = sqrt_r1 * (1.0 - r2)
+    w = sqrt_r1 * r2
+
+    sampled_v0 = vertices[triangles[tri_indices, 0]]
+    sampled_v1 = vertices[triangles[tri_indices, 1]]
+    sampled_v2 = vertices[triangles[tri_indices, 2]]
+
+    points = u[:, None] * sampled_v0 + v[:, None] * sampled_v1 + w[:, None] * sampled_v2
+    logger.info("Sampled %d points (total mesh area: %.4f mm²)", len(points), total_area)
+    return points

src/loader/assembly_tree.py

@@ -0,0 +1,324 @@
+"""Assembly tree extraction from XDE document or shape topology."""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+
+from OCP.Bnd import Bnd_Box
+from OCP.BRepBndLib import BRepBndLib
+from OCP.TDF import TDF_Label, TDF_LabelSequence
+from OCP.TDataStd import TDataStd_Name
+from OCP.TopAbs import TopAbs_COMPOUND, TopAbs_SOLID
+from OCP.TopExp import TopExp_Explorer
+from OCP.TopoDS import TopoDS, TopoDS_Shape
+from OCP.XCAFDoc import XCAFDoc_ShapeTool
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class AssemblyNode:
+    """Node in the assembly tree."""
+    id: str
+    name: str
+    is_assembly: bool = False
+    is_leaf: bool = False
+    shape: TopoDS_Shape | None = None
+    children: list[AssemblyNode] = field(default_factory=list)
+    classification: str = "unknown"  # exterior, interior, unknown
+
+    # Geometry info (populated after tessellation)
+    num_faces: int = 0
+    num_solids: int = 0
+    bounding_box: dict | None = None
+
+    def to_dict(self) -> dict:
+        """Convert to serializable dictionary."""
+        result = {
+            "id": self.id,
+            "name": self.name,
+            "is_assembly": self.is_assembly,
+            "is_leaf": self.is_leaf,
+            "classification": self.classification,
+            "num_faces": self.num_faces,
+            "num_solids": self.num_solids,
+        }
+        if self.bounding_box:
+            result["bounding_box"] = self.bounding_box
+        if self.children:
+            result["children"] = [c.to_dict() for c in self.children]
+        return result
+
+    def iter_leaves(self):
+        """Iterate over all leaf nodes."""
+        if self.is_leaf:
+            yield self
+        for child in self.children:
+            yield from child.iter_leaves()
+
+    def find_by_id(self, node_id: str) -> AssemblyNode | None:
+        """Find a node by ID."""
+        if self.id == node_id:
+            return self
+        for child in self.children:
+            found = child.find_by_id(node_id)
+            if found:
+                return found
+        return None
+
+
+def _get_label_name(label: TDF_Label) -> str:
+    """Extract name from a TDF_Label."""
+    name_attr = TDataStd_Name()
+    if label.FindAttribute(TDataStd_Name.GetID_s(), name_attr):
+        return name_attr.Get().ToExtString()
+    return "unnamed"
+
+
+def _count_solids(shape: TopoDS_Shape) -> int:
+    """Count solid entities in a shape."""
+    count = 0
+    exp = TopExp_Explorer(shape, TopAbs_SOLID)
+    while exp.More():
+        count += 1
+        exp.Next()
+    return count
+
+
+def _compute_bounding_box(shape: TopoDS_Shape) -> dict | None:
+    """Compute the bounding box of a shape."""
+    try:
+        bbox = Bnd_Box()
+        BRepBndLib.Add_s(shape, bbox)
+        if bbox.IsVoid():
+            return None
+        xmin, ymin, zmin, xmax, ymax, zmax = bbox.Get()
+        return {
+            "min": [xmin, ymin, zmin],
+            "max": [xmax, ymax, zmax],
+            "size": [xmax - xmin, ymax - ymin, zmax - zmin],
+            "center": [(xmin + xmax) / 2, (ymin + ymax) / 2, (zmin + zmax) / 2],
+        }
+    except Exception:
+        return None
+
+
+def _build_tree(
+    shape_tool: XCAFDoc_ShapeTool,
+    label: TDF_Label,
+    parent_id: str = "",
+    index: int = 0,
+) -> AssemblyNode:
+    """Recursively build assembly tree from XDE label."""
+    node_id = f"{parent_id}_{index}" if parent_id else str(index)
+    name = _get_label_name(label)
+
+    is_assembly = shape_tool.IsAssembly(label)
+    shape = shape_tool.GetShape(label)
+
+    node = AssemblyNode(
+        id=node_id,
+        name=name,
+        is_assembly=is_assembly,
+        shape=shape if not shape.IsNull() else None,
+    )
+
+    if is_assembly:
+        sub_labels = TDF_LabelSequence()
+        shape_tool.GetComponents(label, sub_labels)
+
+        for i in range(1, sub_labels.Length() + 1):
+            sub_label = sub_labels.Value(i)
+            ref_label = TDF_Label()
+            if shape_tool.GetReferredShape(sub_label, ref_label):
+                child = _build_tree(shape_tool, ref_label, node_id, i)
+            else:
+                child = _build_tree(shape_tool, sub_label, node_id, i)
+            node.children.append(child)
+    else:
+        node.is_leaf = True
+        if node.shape is not None and not node.shape.IsNull():
+            node.num_solids = _count_solids(node.shape)
+            node.bounding_box = _compute_bounding_box(node.shape)
+
+    return node
+
+
+def extract_assembly_tree(shape_tool: XCAFDoc_ShapeTool) -> AssemblyNode:
+    """Extract the full assembly tree from an XDE ShapeTool.
+
+    Args:
+        shape_tool: XCAFDoc_ShapeTool from the loaded document.
+
+    Returns:
+        Root AssemblyNode with the complete tree.
+    """
+    labels = TDF_LabelSequence()
+    shape_tool.GetFreeShapes(labels)
+
+    if labels.Length() == 0:
+        raise RuntimeError("No free shapes found in XDE document")
+
+    if labels.Length() == 1:
+        root = _build_tree(shape_tool, labels.Value(1), "", 0)
+    else:
+        root = AssemblyNode(id="root", name="Assembly", is_assembly=True)
+        for i in range(1, labels.Length() + 1):
+            child = _build_tree(shape_tool, labels.Value(i), "root", i)
+            root.children.append(child)
+
+    leaf_count = sum(1 for _ in root.iter_leaves())
+    logger.info("Assembly tree (XDE): %d leaves, root name='%s'", leaf_count, root.name)
+    return root
+
+
+def _extract_product_names(reader) -> list[str]:
+    """Extract product names from STEP model entities.
+
+    Filters out generic translator names and bare shape type names.
+    """
+    try:
+        from OCP.StepBasic import StepBasic_Product
+        ws = reader.WS()
+        model = ws.Model()
+        skip = {"Open CASCADE STEP translator", "SOLID", "SHELL", "COMPOUND", "COMPSOLID", "WIRE", "EDGE", "VERTEX", "FACE"}
+        names = []
+        for i in range(1, model.NbEntities() + 1):
+            ent = model.Entity(i)
+            if isinstance(ent, StepBasic_Product):
+                name = ent.Name().ToCString() if ent.Name() else None
+                if name and name not in skip and not any(s in name for s in {"Open CASCADE STEP translator"}):
+                    names.append(name)
+        return names
+    except Exception:
+        return []
+
+
+def _extract_xde_label_names(doc) -> list[str]:
+    """Extract shape names from XDE document's Shapes sublabel tree."""
+    try:
+        from OCP.TDF import TDF_ChildIterator
+
+        names = []
+        # First child of Main is the "Shapes" label
+        it = TDF_ChildIterator(doc.Main(), False)
+        if not it.More():
+            return []
+        shapes_label = it.Value()
+
+        # Iterate shape sublabels
+        it2 = TDF_ChildIterator(shapes_label, False)
+        while it2.More():
+            label = it2.Value()
+            name_attr = TDataStd_Name()
+            if label.FindAttribute(TDataStd_Name.GetID_s(), name_attr):
+                raw = name_attr.Get().ToExtString()
+                # Skip translator meta-names
+                if "Open CASCADE STEP translator" not in raw:
+                    names.append(raw)
+            it2.Next()
+
+        return names
+    except Exception:
+        return []
+
+
+def _load_names_json(step_path) -> list[str]:
+    """Load part names from a .names.json sidecar file."""
+    import json
+    from pathlib import Path
+
+    names_path = Path(str(step_path) + ".names.json")
+    if not names_path.exists():
+        # Also try replacing extension
+        names_path = Path(step_path).with_suffix(".names.json")
+    if not names_path.exists():
+        return []
+
+    try:
+        with open(names_path) as f:
+            names = json.load(f)
+        if isinstance(names, list):
+            return names
+    except Exception:
+        pass
+    return []
+
+
+def extract_assembly_from_shape(
+    shape: TopoDS_Shape,
+    name: str = "Assembly",
+    reader=None,
+    doc=None,
+    step_path=None,
+    id_prefix: str = "",
+) -> AssemblyNode:
+    """Fallback: extract assembly tree from shape topology.
+
+    Enumerates solids from the shape compound to create leaf nodes.
+    Tries to map names from: XDE labels, or STEP products.
+
+    Args:
+        shape: The TopoDS_Shape to decompose.
+        name: Root node name.
+        reader: Optional STEPControl_Reader for product name extraction.
+        doc: Optional TDocStd_Document for XDE label name extraction.
+        step_path: Path to the STEP file.
+        id_prefix: Prefix for node IDs to avoid collisions between files.
+
+    Returns:
+        Root AssemblyNode with solids as leaves.
+    """
+    part_names: list[str] = []
+
+    # Source 1: STEP product entities (most reliable for named parts)
+    if reader is not None:
+        product_names = _extract_product_names(reader)
+        if product_names:
+            part_names = product_names
+            logger.info("Found %d part names from STEP products", len(part_names))
+
+    # Source 2: XDE document labels (fallback)
+    if not part_names and doc is not None:
+        xde_names = _extract_xde_label_names(doc)
+        # Filter out generic shape type names
+        skip = {"SOLID", "SHELL", "COMPOUND", "COMPSOLID", "WIRE", "EDGE", "VERTEX", "FACE"}
+        xde_names = [n for n in xde_names if n not in skip]
+        if xde_names:
+            part_names = xde_names
+            logger.info("Found %d part names from XDE labels", len(part_names))
+
+    root_id = f"{id_prefix}0" if id_prefix else "0"
+    root = AssemblyNode(id=root_id, name=name, is_assembly=True, shape=shape)
+
+    solid_exp = TopExp_Explorer(shape, TopAbs_SOLID)
+    idx = 0
+    while solid_exp.More():
+        solid = TopoDS.Solid_s(solid_exp.Current())
+        if idx < len(part_names):
+            solid_name = part_names[idx]
+        else:
+            solid_name = f"Part_{idx + 1}"
+
+        node = AssemblyNode(
+            id=f"{root_id}_{idx + 1}",
+            name=solid_name,
+            is_leaf=True,
+            shape=solid,
+            num_solids=1,
+            bounding_box=_compute_bounding_box(solid),
+        )
+        root.children.append(node)
+        idx += 1
+        solid_exp.Next()
+
+    leaf_count = len(root.children)
+    if leaf_count == 0:
+        root.is_leaf = True
+        root.is_assembly = False
+        root.num_solids = 0
+        root.bounding_box = _compute_bounding_box(shape)
+    else:
+        logger.info("Assembly tree (shape): %d solids from topology", leaf_count)
+
+    return root

src/loader/step_loader.py

@@ -0,0 +1,169 @@
+"""STEP file loading with XDE support and Shape Healing."""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from OCP.IFSelect import IFSelect_RetDone
+from OCP.STEPCAFControl import STEPCAFControl_Reader
+from OCP.STEPControl import STEPControl_Reader
+from OCP.ShapeFix import ShapeFix_Shape, ShapeFix_ShapeTolerance
+from OCP.TCollection import TCollection_ExtendedString
+from OCP.TDocStd import TDocStd_Document
+from OCP.TopoDS import TopoDS_Shape
+from OCP.XCAFApp import XCAFApp_Application
+from OCP.XCAFDoc import XCAFDoc_ShapeTool
+from OCP.TDF import TDF_LabelSequence
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class StepFileInfo:
+    """Metadata about a loaded STEP file."""
+    path: Path
+    shape: TopoDS_Shape
+    doc: TDocStd_Document | None = None
+    shape_tool: XCAFDoc_ShapeTool | None = None
+    reader: STEPControl_Reader | None = None
+    protocol: str = ""
+    unit: str = "mm"
+    num_roots: int = 0
+    extra: dict = field(default_factory=dict)
+
+
+def _heal_shape(shape: TopoDS_Shape, tolerance: float = 0.01) -> TopoDS_Shape:
+    """Apply shape healing to fix geometry issues."""
+    fixer = ShapeFix_Shape(shape)
+    fixer.SetPrecision(tolerance)
+    fixer.SetMaxTolerance(tolerance * 10)
+    fixer.Perform()
+
+    tol_fixer = ShapeFix_ShapeTolerance()
+    tol_fixer.SetTolerance(fixer.Shape(), tolerance)
+
+    logger.info("Shape healing completed")
+    return fixer.Shape()
+
+
+def _detect_protocol(reader: STEPControl_Reader) -> str:
+    """Detect AP protocol from STEP file."""
+    try:
+        ws = reader.WS()
+        model = ws.Model()
+        if model is not None:
+            header = str(model.Header()) if hasattr(model, "Header") else ""
+            for ap in ("AP214", "AP203", "AP242"):
+                if ap.lower() in header.lower() or ap in header:
+                    return ap
+    except Exception:
+        pass
+    return "unknown"
+
+
+def _load_xde(file_path: Path) -> tuple[TDocStd_Document, XCAFDoc_ShapeTool, str]:
+    """Load STEP into XDE document and return (doc, shape_tool, protocol)."""
+    app = XCAFApp_Application.GetApplication_s()
+    doc = TDocStd_Document(TCollection_ExtendedString("MDTV-XCAF"))
+    app.InitDocument(doc)
+
+    xde_reader = STEPCAFControl_Reader()
+    xde_reader.SetNameMode(True)
+    xde_reader.SetColorMode(True)
+    xde_reader.SetLayerMode(True)
+
+    status = xde_reader.ReadFile(str(file_path))
+    if status != IFSelect_RetDone:
+        raise RuntimeError(f"Failed to read STEP file: {file_path} (status={status})")
+
+    protocol = _detect_protocol(xde_reader.Reader())
+
+    if not xde_reader.Transfer(doc):
+        raise RuntimeError(f"Failed to transfer STEP data: {file_path}")
+
+    shape_tool = XCAFDoc_ShapeTool.Set_s(doc.Main())
+    return doc, shape_tool, protocol
+
+
+def _load_basic(file_path: Path) -> tuple[TopoDS_Shape, str, int, STEPControl_Reader]:
+    """Fallback: load with basic STEPControl_Reader."""
+    reader = STEPControl_Reader()
+    status = reader.ReadFile(str(file_path))
+    if status != IFSelect_RetDone:
+        raise RuntimeError(f"Failed to read STEP file: {file_path}")
+
+    protocol = _detect_protocol(reader)
+    num_roots = reader.NbRootsForTransfer()
+    reader.TransferRoots()
+    shape = reader.OneShape()
+    return shape, protocol, num_roots, reader
+
+
+def load_step(file_path: str | Path, heal: bool = True) -> StepFileInfo:
+    """Load a STEP file, trying XDE reader first then falling back to basic.
+
+    Args:
+        file_path: Path to the .stp/.step file.
+        heal: Whether to apply shape healing.
+
+    Returns:
+        StepFileInfo with shape, XDE document (if available), and metadata.
+    """
+    file_path = Path(file_path)
+    if not file_path.exists():
+        raise FileNotFoundError(f"STEP file not found: {file_path}")
+
+    logger.info("Loading STEP file: %s (%.1f MB)", file_path, file_path.stat().st_size / 1e6)
+
+    doc = None
+    shape_tool = None
+    shape = None
+    basic_reader = None
+    protocol = "unknown"
+    num_roots = 0
+
+    # Try XDE reader for assembly structure
+    try:
+        doc, shape_tool, protocol = _load_xde(file_path)
+
+        labels = TDF_LabelSequence()
+        shape_tool.GetFreeShapes(labels)
+        num_roots = labels.Length()
+        logger.info("XDE reader: %d free shape(s), protocol=%s", num_roots, protocol)
+
+        if num_roots > 0:
+            if num_roots == 1:
+                shape = shape_tool.GetShape(labels.Value(1))
+            else:
+                from OCP.BRep import BRep_Builder
+                from OCP.TopoDS import TopoDS_Compound
+                builder = BRep_Builder()
+                compound = TopoDS_Compound()
+                builder.MakeCompound(compound)
+                for i in range(1, num_roots + 1):
+                    builder.Add(compound, shape_tool.GetShape(labels.Value(i)))
+                shape = compound
+    except Exception as e:
+        logger.warning("XDE reader failed: %s", e)
+
+    # Fallback to basic reader if XDE produced no shapes
+    if shape is None or shape.IsNull():
+        logger.info("Falling back to basic STEPControl_Reader")
+        shape, protocol, num_roots, basic_reader = _load_basic(file_path)
+
+    if shape.IsNull():
+        raise RuntimeError(f"Empty shape in STEP file: {file_path}")
+
+    if heal:
+        shape = _heal_shape(shape)
+
+    return StepFileInfo(
+        path=file_path,
+        shape=shape,
+        doc=doc,
+        shape_tool=shape_tool,
+        reader=basic_reader,
+        protocol=protocol,
+        num_roots=num_roots,
+    )

src/mcp/server.py

@@ -0,0 +1,30 @@
+"""MCP server for CAD review tools."""
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+# MCP server setup - this will be a FastMCP-based server
+# that exposes tools for interacting with the CAD review system
+
+try:
+    from mcp.server.fastmcp import FastMCP
+
+    mcp = FastMCP("cad-review")
+
+    from src.mcp.tools import register_tools
+    register_tools(mcp)
+
+except ImportError:
+    logger.info("MCP SDK not available - MCP server disabled")
+    mcp = None
+
+
+def run_mcp_server():
+    """Run the MCP server."""
+    if mcp is None:
+        raise RuntimeError("MCP SDK not installed. Install with: pip install mcp")
+    mcp.run()

src/mcp/tools.py

@@ -0,0 +1,155 @@
+"""MCP tool definitions for CAD review."""
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+
+def register_tools(mcp):
+    """Register all MCP tools."""
+
+    @mcp.tool()
+    def get_assembly_tree(job_id: str) -> str:
+        """Get the assembly tree structure for a loaded STEP file.
+
+        Args:
+            job_id: The job ID from uploading a STEP file.
+        """
+        from src.api.router import get_manager
+        manager = get_manager()
+        job = manager.get_job(job_id)
+        if job is None:
+            return f"Error: Job not found: {job_id}"
+        if job.assembly_tree is None:
+            return f"Error: Assembly tree not available yet (status: {job.status})"
+
+        import json
+        return json.dumps(job.assembly_tree.to_dict(), indent=2)
+
+    @mcp.tool()
+    def get_part_info(job_id: str, part_id: str) -> str:
+        """Get detailed information about a specific part.
+
+        Args:
+            job_id: The job ID.
+            part_id: The part ID from the assembly tree.
+        """
+        from src.api.router import get_manager
+        manager = get_manager()
+        job = manager.get_job(job_id)
+        if job is None:
+            return f"Error: Job not found: {job_id}"
+        if job.assembly_tree is None:
+            return "Error: Assembly tree not available yet"
+
+        node = job.assembly_tree.find_by_id(part_id)
+        if node is None:
+            return f"Error: Part not found: {part_id}"
+
+        import json
+        info = node.to_dict()
+        mesh = job.part_meshes.get(part_id)
+        if mesh:
+            info["mesh_vertices"] = len(mesh.vertices)
+            info["mesh_triangles"] = len(mesh.triangles)
+        return json.dumps(info, indent=2)
+
+    @mcp.tool()
+    def measure_part_distance(job_id: str, part_a_id: str, part_b_id: str) -> str:
+        """Measure the minimum distance between two parts.
+
+        Args:
+            job_id: The job ID.
+            part_a_id: ID of the first part.
+            part_b_id: ID of the second part.
+        """
+        from src.api.router import get_manager
+        from src.geometry.measurement import measure_distance
+
+        manager = get_manager()
+        job = manager.get_job(job_id)
+        if job is None:
+            return f"Error: Job not found: {job_id}"
+        if job.assembly_tree is None:
+            return "Error: Assembly tree not available yet"
+
+        node_a = job.assembly_tree.find_by_id(part_a_id)
+        node_b = job.assembly_tree.find_by_id(part_b_id)
+
+        if node_a is None or node_a.shape is None:
+            return f"Error: Part not found or has no shape: {part_a_id}"
+        if node_b is None or node_b.shape is None:
+            return f"Error: Part not found or has no shape: {part_b_id}"
+
+        result = measure_distance(node_a.shape, node_b.shape)
+
+        import json
+        return json.dumps({
+            "distance_mm": result.distance_mm,
+            "point_a": result.point_a,
+            "point_b": result.point_b,
+        }, indent=2)
+
+    @mcp.tool()
+    def check_compliance_rules(job_id: str) -> str:
+        """Run compliance checks on the loaded assembly.
+
+        Args:
+            job_id: The job ID.
+        """
+        from src.api.router import get_manager
+        from src.compliance.kmvss_checker import check_compliance
+        from src.compliance.rule_loader import load_rules
+
+        manager = get_manager()
+        job = manager.get_job(job_id)
+        if job is None:
+            return f"Error: Job not found: {job_id}"
+        if job.assembly_tree is None:
+            return "Error: Assembly tree not available yet"
+
+        rules_path = Path(__file__).parent.parent.parent / "config" / "kmvss_rules.yaml"
+        rules = load_rules(rules_path)
+        results = check_compliance(rules, job.assembly_tree)
+
+        import json
+        return json.dumps([{
+            "rule": r.rule_name,
+            "passed": r.passed,
+            "message": r.message,
+            "severity": r.severity,
+        } for r in results], indent=2)
+
+    @mcp.tool()
+    def get_collision_report(job_id: str) -> str:
+        """Get a collision/proximity report for all parts.
+
+        Args:
+            job_id: The job ID.
+        """
+        from src.api.router import get_manager
+        from src.geometry.measurement import scan_proximity
+
+        manager = get_manager()
+        job = manager.get_job(job_id)
+        if job is None:
+            return f"Error: Job not found: {job_id}"
+        if job.assembly_tree is None:
+            return "Error: Assembly tree not available yet"
+
+        parts = []
+        for leaf in job.assembly_tree.iter_leaves():
+            if leaf.shape is not None and not leaf.shape.IsNull():
+                parts.append({"id": leaf.id, "name": leaf.name, "shape": leaf.shape})
+
+        results = scan_proximity(parts)
+
+        import json
+        return json.dumps([{
+            "part_a": r.part_a_name,
+            "part_b": r.part_b_name,
+            "distance_mm": r.min_distance_mm,
+            "status": r.status,
+        } for r in results], indent=2)