"""CDSL runtime adapter driven exclusively by profile operation contracts.""" from __future__ import annotations from copy import deepcopy import json import math from pathlib import Path from typing import Any from app.cad_agent.domain.operation_contract import ( OperationContractError, canonical_hash, is_authoring_schema_closed, validate_operation_contract, ) from app.cad_agent.ports import AdapterUnavailable from app.services.engine_service import load_engine, topology_snapshot, validate_cdsl, validate_cdsl_shape from app.services.render_bundle import RenderBundleError, render_checkpoint from app.settings import Settings from vendor.cdsl_preview_runtime import step_to_glb class RuntimeAdapterError(RuntimeError): pass class RuntimeServiceUnavailable(AdapterUnavailable): """A renderer or artifact service outage that must not spend a repair.""" class ProfileCadRuntime: """Adapter that owns engine imports; application code sees only its port.""" def __init__(self, settings: Settings) -> None: self.settings = settings self.engine = load_engine(settings) schema_path = Path(str(self.engine.__file__)).with_name("profile_schema.json") try: self._profile = json.loads(schema_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise RuntimeAdapterError("RUNTIME_CONTRACT_INVALID: profile schema is unavailable") from error profile_contracts = self._profile.get("operation_contracts") if not isinstance(profile_contracts, dict): raise RuntimeAdapterError("RUNTIME_CONTRACT_INVALID: profile has no operation contracts") declared = {str(item) for item in profile_contracts} registered = {str(item) for item in getattr(self.engine, "SUPPORTED_ATOMIC_IDS", ())} if declared != registered: raise RuntimeAdapterError("RUNTIME_CONTRACT_INVALID: runtime and profile operation registries disagree") self._contracts: dict[str, dict[str, Any]] = {} for atomic_id, contract in profile_contracts.items(): if not isinstance(contract, dict): raise RuntimeAdapterError("RUNTIME_CONTRACT_INVALID: operation contract is not an object") # A legacy opaque operation schema is still executable internally, # but cannot enter the sole LLM-facing Authoring protocol. if not is_authoring_schema_closed(contract.get("author_params_schema")): continue try: validate_operation_contract(contract) except OperationContractError as error: raise RuntimeAdapterError(f"RUNTIME_CONTRACT_INVALID: {error}") from error self._contracts[str(atomic_id)] = contract if not self._contracts: raise RuntimeAdapterError("RUNTIME_CONTRACT_INVALID: profile has no strict authoring operations") def supported_atomic_ids(self) -> tuple[str, ...]: return tuple(sorted(self._contracts)) def operation_contract(self, atomic_id: str) -> dict[str, Any]: contract = self._contracts.get(atomic_id) if not isinstance(contract, dict): raise RuntimeAdapterError(f"RUNTIME_CONTRACT_INVALID: unknown operation {atomic_id}") result = deepcopy(contract) result["contract_hash"] = canonical_hash(contract) result["registry_revision"] = str(self._profile.get("schema_version") or "") return result def compile_authoring(self, document: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: """Compile model-facing Authoring CDSL using server-owned contracts.""" from app.cad_agent.application.authoring_compiler import AuthoringCompiler from app.cad_agent.application.authoring_compiler import AuthoringCompileError runtime, audit = AuthoringCompiler(self.operation_contract).compile(document) try: self._validate_finite_tree(runtime) validate_cdsl_shape(runtime, self.engine) except Exception as error: raise AuthoringCompileError( "RUNTIME_CONTRACT_INVALID", f"compiled Runtime CDSL is not executable: {error}", ) from error return runtime, audit def build_checkpoint(self, cdsl: dict[str, Any], output_dir: str, task_id: str, revision_id: str) -> dict[str, Any]: """Build the exact geometry checkpoint required by every DAG node. Rendering is deliberately absent here: a renderer outage must never make a geometrically valid atomic feature fail its local acceptance. """ root = Path(output_dir) root.mkdir(parents=True, exist_ok=True) cdsl_copy = deepcopy(cdsl) cdsl_copy["part_id"] = task_id step_path = root / "model.step" self._write_json(root / "model.cdsl.json", cdsl_copy) try: self._validate_finite_tree(cdsl_copy) validate_cdsl(cdsl_copy, self.engine) engine_result = self.engine.run_cdsl_only(cdsl_copy, step_path) if str(engine_result.get("engine") or "") != "cdsl_only": raise RuntimeAdapterError("engine did not execute CDSL-only rebuild") health = self._health(engine_result, step_path) topology = topology_snapshot(engine_result, task_id=task_id, revision_id=revision_id) self._write_json(root / "model.topology.json", topology) report = {"engine_result": engine_result, "preview": {}, "health": health, "render_manifest": {}} self._write_json(root / "rebuild-report.json", report) return {"health": health, "topology": topology, "report": report, "render_manifest": {}, "paths": {"cdsl": "model.cdsl.json", "step": "model.step", "topology": "model.topology.json", "report": "rebuild-report.json"}} except OSError: raise except RuntimeAdapterError: raise except Exception as error: raise RuntimeAdapterError(f"RUNTIME_EXECUTION_FAILURE: {error}") from error def create_preview(self, output_dir: str) -> dict[str, Any]: """Create an optional GLB preview for a completed checkpoint.""" root = Path(output_dir) step_path = root / "model.step" glb_path = root / "model.glb" preview = step_to_glb(step_path, glb_path) report_path = root / "rebuild-report.json" report = json.loads(report_path.read_text(encoding="utf-8")) if report_path.is_file() else {} report["preview"] = preview self._write_json(report_path, report) return {"preview": preview, "path": "model.glb"} def render_bundle(self, output_dir: str) -> dict[str, Any]: root = Path(output_dir) manifest = render_checkpoint(self.settings, step_path=root / "model.step", output_dir=root / "renders") report_path = root / "rebuild-report.json" report = json.loads(report_path.read_text(encoding="utf-8")) if report_path.is_file() else {} report["render_manifest"] = manifest self._write_json(report_path, report) return manifest def rebuild(self, cdsl: dict[str, Any], output_dir: str, task_id: str, revision_id: str) -> dict[str, Any]: """Build one complete checkpoint and derive all publishable artifacts.""" built = self.build_checkpoint(cdsl, output_dir, task_id, revision_id) root = Path(output_dir) try: preview = self.create_preview(output_dir) manifest = self.render_bundle(output_dir) report = json.loads((root / "rebuild-report.json").read_text(encoding="utf-8")) return {**built, "report": report, "preview": preview["preview"], "render_manifest": manifest, "paths": {**built["paths"], "glb": "model.glb", "render_manifest": "renders/render-manifest.json"}} except OSError: # Artifact writes are a recoverable infrastructure outage. Let the # application handler park the same candidate stage for replay. raise except RenderBundleError as error: raise RuntimeServiceUnavailable(f"RENDER_SERVICE_UNAVAILABLE: {error}") from error except Exception as error: raise RuntimeAdapterError(f"RUNTIME_EXECUTION_FAILURE: {error}") from error def rebuild_best_effort( self, cdsl: dict[str, Any], output_dir: str, task_id: str, revision_id: str, ) -> tuple[dict[str, Any], list[dict[str, Any]]]: """Build independent feature prefixes without discarding prior geometry. A full CDSL document can contain several features even though the action protocol normally appends one at a time. Engine validation is all-or-nothing, so replay features in source order and retain each executable feature. A failed feature is reported as a diagnostic; it does not erase already executable geometry or prevent later, independent features from being tried. """ features = cdsl.get("features") if isinstance(cdsl.get("features"), list) else [] accepted: list[dict[str, Any]] = [] accepted_ids: set[str] = set() failures: list[dict[str, Any]] = [] latest: dict[str, Any] | None = None for index, feature in enumerate(features): if not isinstance(feature, dict): failures.append({"feature_index": index, "feature_id": "", "message": "Feature must be an object."}) continue feature_id = str(feature.get("id") or f"feature_{index + 1:03d}") dependencies = [str(value) for value in feature.get("depends_on") or () if isinstance(value, str)] unavailable = [value for value in dependencies if value not in accepted_ids] if unavailable: failures.append({ "code": "SELECTOR_DEPENDENCY_UNAVAILABLE" if feature.get("selectors") else "DEPENDENCY_UNAVAILABLE", "feature_index": index, "feature_id": feature_id, "message": "Feature was skipped because an earlier dependency did not execute.", "dependencies": unavailable, }) continue candidate = self._feature_subset(cdsl, [*accepted, feature]) try: latest = self.rebuild(candidate, output_dir, task_id, revision_id) except (OSError, AdapterUnavailable): # Preview/render/storage failures are service faults. Preserve # the staging input and let the workflow replay this exact # document without spending an Authoring repair. raise except Exception as error: failures.append(self._build_failure(index, feature_id, feature, error)) continue accepted.append(deepcopy(feature)) accepted_ids.add(feature_id) if not accepted: if failures: return {}, failures raise RuntimeAdapterError("RUNTIME_EXECUTION_FAILURE: CDSL document has no executable features") # A failed later attempt may have left partial files in the stage. # Rebuild the retained feature set once so all published artifacts are # guaranteed to describe the same successful checkpoint. latest = self.rebuild(self._feature_subset(cdsl, accepted), output_dir, task_id, revision_id) return {**latest, "executed_feature_ids": sorted(accepted_ids)}, failures @staticmethod def _build_failure(index: int, feature_id: str, feature: dict[str, Any], error: Exception) -> dict[str, Any]: diagnostic = getattr(error, "diagnostic", None) message = str(getattr(diagnostic, "message", "") or error) diagnostic_code = str(getattr(diagnostic, "code", "") or "") if not diagnostic_code: diagnostic_code = next(( code for code in ( "selector_output_role_not_found", "selector_geometry_mismatch", "selector_output_role_ambiguous", "selector_relation_non_unique", "selector_output_role_owner_required", "selector_output_role_active_body_required", "selector_owner_required", "selector_output_role_mixed_evidence", "unsupported_output_role_selector", "invalid_output_role_selector", ) if code in message ), "") selector_code = { "selector_output_role_not_found": "SELECTOR_NOT_FOUND", "selector_geometry_mismatch": "SELECTOR_NOT_FOUND", "selector_output_role_ambiguous": "SELECTOR_AMBIGUOUS", "selector_relation_non_unique": "SELECTOR_AMBIGUOUS", "selector_output_role_owner_required": "SELECTOR_DEPENDENCY_UNAVAILABLE", "selector_output_role_active_body_required": "SELECTOR_DEPENDENCY_UNAVAILABLE", "selector_owner_required": "SELECTOR_DEPENDENCY_UNAVAILABLE", "selector_output_role_mixed_evidence": "SELECTOR_KIND_MISMATCH", "unsupported_output_role_selector": "SELECTOR_KIND_MISMATCH", "invalid_output_role_selector": "SELECTOR_KIND_MISMATCH", }.get(diagnostic_code, "ENGINE_EXECUTION_FAILED") provenance = getattr(error, "selector_resolutions", None) return { "code": selector_code, "runtime_code": diagnostic_code or "execution_failed", "feature_index": index, "feature_id": feature_id, "atomic_id": str(feature.get("atomic_id") or ""), "input_summary": {"params": sorted((feature.get("params") or {}).keys()), "selector_count": len(feature.get("selectors") or [])}, "message": message[:1000], "selector_provenance": provenance if isinstance(provenance, list) else [], } @staticmethod def _feature_subset(cdsl: dict[str, Any], features: list[dict[str, Any]]) -> dict[str, Any]: document = deepcopy(cdsl) document["features"] = deepcopy(features) geometry = document.get("geometry") if isinstance(document.get("geometry"), dict) else {} sketches = geometry.get("sketches") if isinstance(geometry.get("sketches"), list) else [] sketch_ids = {str(feature.get("sketch_id") or "") for feature in features} document["geometry"] = { **geometry, "sketches": [deepcopy(sketch) for sketch in sketches if isinstance(sketch, dict) and str(sketch.get("id") or "") in sketch_ids], } return document @staticmethod def _validate_finite_tree(value: Any, path: str = "$") -> None: if isinstance(value, float) and not math.isfinite(value): raise RuntimeAdapterError(f"RUNTIME_CONTRACT_INVALID: non-finite number at {path}") if isinstance(value, dict): for key, child in value.items(): ProfileCadRuntime._validate_finite_tree(child, f"{path}.{key}") elif isinstance(value, list): for index, child in enumerate(value): ProfileCadRuntime._validate_finite_tree(child, f"{path}[{index}]") @staticmethod def _write_json(path: Path, payload: dict[str, Any]) -> None: path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") @staticmethod def _health(engine_result: dict[str, Any], step_path: Path, glb_path: Path | None = None) -> dict[str, Any]: bbox = engine_result.get("bbox_mm") if isinstance(engine_result.get("bbox_mm"), dict) else {} minimum, maximum = bbox.get("min"), bbox.get("max") if not isinstance(minimum, list) or not isinstance(maximum, list) or len(minimum) != 3 or len(maximum) != 3 or not step_path.is_file() or (glb_path is not None and not glb_path.is_file()): raise RuntimeAdapterError("rebuild did not produce complete deterministic artifacts") return {"bbox_mm": {"min": [float(item) for item in minimum], "max": [float(item) for item in maximum], "dimensions": [float(maximum[index]) - float(minimum[index]) for index in range(3)]}, "volume_mm3": float(engine_result["volume_mm3"]), "solid_count": int(engine_result["solid_count"]), "feature_count": len(engine_result.get("feature_results") or [])}