"""Append-only CDSL materialisation for the autonomous authoring loop.""" from __future__ import annotations from copy import deepcopy from hashlib import sha256 import json from typing import Any class CdslFragmentError(ValueError): """A fragment cannot safely be applied to the current CDSL state.""" class AutonomousFragmentError(CdslFragmentError): """A free-form candidate violates the autonomous append-only boundary.""" def cdsl_sha256(cdsl: dict[str, Any] | None) -> str: value = cdsl or {"geometry": {"sketches": []}, "features": []} return sha256(json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() def materialize_fragment(base_cdsl: dict[str, Any] | None, fragment: dict[str, Any]) -> dict[str, Any]: """Create the complete CDSL document that will be rebuilt from scratch.""" if base_cdsl is None: document: dict[str, Any] = { "schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", "part_id": "agent_preflight", "geometry": {"sketches": []}, "features": [], } else: document = deepcopy(base_cdsl) geometry = document.setdefault("geometry", {}) sketches = geometry.setdefault("sketches", []) if isinstance(geometry, dict) else None features = document.setdefault("features", []) if not isinstance(sketches, list) or not isinstance(features, list): raise CdslFragmentError("Base CDSL has invalid geometry collections") sketches.extend(deepcopy(fragment["add_sketches"])) features.extend(deepcopy(fragment["add_features"])) return document def selector_bindings(engine_result: dict[str, Any], *, node_id: str, snapshot_id: str) -> dict[str, Any]: """Persist runtime selector choices as build evidence.""" values = [] for resolution in engine_result.get("selector_resolution") or (): if not isinstance(resolution, dict): continue selector = resolution.get("selector") if isinstance(resolution.get("selector"), dict) else {} values.append({ "consumer_node_id": node_id, "consumer_feature_id": str(resolution.get("feature_id") or ""), "source_snapshot_id": str(selector.get("snapshot_id") or snapshot_id), "kind": str(selector.get("kind") or ""), "owner_feature_id": str(selector.get("owner_feature_id") or ""), "stable_id": str(selector.get("stable_id") or ""), "geometry": deepcopy(selector.get("geometry") or {}), "status": str(resolution.get("status") or ""), "candidates": deepcopy(list(resolution.get("candidates") or [])), "score": resolution.get("score"), "selected": deepcopy(resolution.get("selected") or resolution.get("record") or {}), }) return {"schema_version": "cad.selector-bindings.v1", "node_id": node_id, "bindings": values} def _autonomous_id(prefix: str, used: set[str]) -> str: index = 1 while True: candidate = f"{prefix}_{index:03d}" if candidate not in used: used.add(candidate) return candidate index += 1 def autonomous_selector_tokens(snapshot: dict[str, Any] | None) -> dict[str, dict[str, Any]]: """Make opaque, revision-scoped selector tokens from executable topology.""" if not isinstance(snapshot, dict): return {} snapshot_id = str(snapshot.get("snapshot_id") or "") if not snapshot_id: return {} values: dict[str, dict[str, Any]] = {} for record in snapshot.get("records") or (): if not isinstance(record, dict) or not record.get("executable"): continue record_id = str(record.get("record_id") or "") kind = str(record.get("kind") or "") if not record_id or kind not in {"face", "edge", "vertex", "plane", "axis", "body"}: continue token = "sel_" + sha256(f"{snapshot_id}|{record_id}".encode("utf-8")).hexdigest()[:16] geometry = deepcopy(record.get("geometry") or {}) owners = record.get("owner_feature_ids") or [record.get("feature_id") or ""] values[token] = { "token": token, "kind": kind, "geometry": geometry, "selector": { "kind": kind, "stable_id": record_id, "owner_feature_id": str(owners[0] or ""), "geometry": geometry, "source": "runtime_snapshot", "snapshot_id": snapshot_id, "confidence": 1.0, }, } return values def _compact_prompt_geometry(geometry: dict[str, Any]) -> dict[str, Any]: """Keep only selector-choice facts useful to an author model. Runtime snapshots also carry adjacency signatures, curve endpoints and other diagnostic detail. Those fields are required for deterministic engine work, but repeatedly placing them in an LLM prompt is expensive and does not help choose an opaque token. Full records remain available on disk and through targeted measurement tools. """ useful = ( "bbox_mm", "center_mm", "normal", "plane_normal", "plane_offset_mm", "surface_type", "curve_type", "radius_mm", "length_mm", "area_mm2", "volume_mm3", "solid_count", ) return {key: deepcopy(geometry[key]) for key in useful if key in geometry} def autonomous_candidate_prompt_tokens(tokens: dict[str, dict[str, Any]], *, kind: str = "") -> list[dict[str, Any]]: """Return compact safe token fields, never persistent selector IDs.""" return [ {"token": token, "kind": value["kind"], "geometry": _compact_prompt_geometry(value["geometry"])} for token, value in sorted(tokens.items()) if not kind or value["kind"] == kind ] def _fragment_lists(fragment: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: if not isinstance(fragment, dict): raise AutonomousFragmentError("fragment_json must decode to a JSON object") # Ordinary tool-call providers occasionally lift the unambiguous selector # array one level out of a single-feature payload. Accept that shorthand # only when it can be moved to exactly one feature without choosing or # changing any selector ourselves. allowed = {"sketch", "feature", "sketches", "features", "add_sketches", "add_features", "selector_tokens", "revolve_axis"} unknown = sorted(set(fragment) - allowed) if unknown: raise AutonomousFragmentError("fragment_json may contain only sketch(es) and feature(s); unexpected: " + ", ".join(unknown)) sketches_raw = fragment.get("sketches", fragment.get("add_sketches", [])) features_raw = fragment.get("features", fragment.get("add_features", [])) if "sketch" in fragment: if sketches_raw: raise AutonomousFragmentError("Use either sketch or sketches, not both") sketches_raw = [fragment["sketch"]] if "feature" in fragment: if features_raw: raise AutonomousFragmentError("Use either feature or features, not both") features_raw = [fragment["feature"]] if not isinstance(sketches_raw, list) or not all(isinstance(item, dict) for item in sketches_raw): raise AutonomousFragmentError("fragment sketches must be an array of objects") if not isinstance(features_raw, list) or not features_raw or not all(isinstance(item, dict) for item in features_raw): raise AutonomousFragmentError("fragment features must be a non-empty array of objects") sketches = deepcopy(sketches_raw) features = deepcopy(features_raw) top_level_tokens = fragment.get("selector_tokens") if top_level_tokens is not None: if len(features) != 1: raise AutonomousFragmentError("top-level selector_tokens are allowed only with exactly one feature") if "selector_tokens" in features[0]: raise AutonomousFragmentError("selector_tokens must appear either at fragment top level or feature level, not both") features[0]["selector_tokens"] = deepcopy(top_level_tokens) return sketches, features def _move_equivalent_field( value: dict[str, Any], *, source: str, target: str, location: str, fixes: list[dict[str, str]], ) -> None: """Move a lossless spelling alias without choosing any CAD geometry.""" if source not in value: return if target in value: if value[source] != value[target]: raise AutonomousFragmentError( f"CONFLICTING_PARAMETER_ALIASES at {location}: both {source} and {target} were supplied with different values" ) value.pop(source) fixes.append({"path": location, "from": source, "to": target, "action": "deduplicated_equivalent"}) return value[target] = value.pop(source) fixes.append({"path": location, "from": source, "to": target, "action": "renamed_equivalent"}) def _move_axis_component( params: dict[str, Any], axis: dict[str, Any], *, source: str, target: str, location: str, fixes: list[dict[str, str]], ) -> None: """Move an explicit top-level axis alias into the canonical axis object.""" if source not in params: return value = params.pop(source) if target in axis and axis[target] != value: raise AutonomousFragmentError( f"CONFLICTING_PARAMETER_ALIASES at {location}: both {source} and axis.{target} were supplied with different values" ) if target in axis: fixes.append({"path": location, "from": source, "to": f"axis.{target}", "action": "deduplicated_equivalent"}) return axis[target] = value fixes.append({"path": location, "from": source, "to": f"axis.{target}", "action": "renamed_equivalent"}) def _normalize_axis_mapping(axis: dict[str, Any], *, location: str, fixes: list[dict[str, str]]) -> None: """Normalize only lossless aliases used inside an already explicit axis.""" for source in ("origin", "point_mm", "axis_origin_mm", "axis_point_mm"): _move_equivalent_field(axis, source=source, target="origin_mm", location=location, fixes=fixes) for source in ("axis_dir", "axis_direction", "dir"): _move_equivalent_field(axis, source=source, target="direction", location=location, fixes=fixes) def _lift_feature_local_sketches(fragment: dict[str, Any], *, fixes: list[dict[str, str]]) -> None: """Accept the common one-feature/one-sketch nesting without choosing geometry. The public fragment grammar owns one ordered sketch list and one ordered feature list. Some tool-call models naturally nest each sketch below its feature. That representation is losslessly transformable only when every feature supplies exactly one sketch and no root sketch collection exists. """ if any(key in fragment for key in ("sketch", "sketches", "add_sketches")): return features = fragment.get("features", fragment.get("add_features")) if not isinstance(features, list) or not features or not all(isinstance(item, dict) for item in features): return lifted: list[dict[str, Any]] = [] for index, feature in enumerate(features): nested = feature.get("sketches", feature.get("sketch")) if isinstance(nested, dict): sketches = [nested] elif isinstance(nested, list): sketches = nested else: return if len(sketches) != 1 or not isinstance(sketches[0], dict): return feature.pop("sketch", None) feature.pop("sketches", None) lifted.append(sketches[0]) fixes.append({"path": f"features[{index}]", "from": "feature-local sketch", "to": "sketches[]", "action": "lifted_equivalent"}) fragment["sketches"] = lifted def normalize_autonomous_fragment(fragment: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, str]]]: """Normalize only explicitly equivalent author spellings. Normalization is deliberately narrow. It accepts common CAD vocabulary where the target runtime field has identical units and meaning, while refusing inputs that would need a guessed profile, direction, selector, coordinate system, or topology decision. The original fragment and every applied fix are retained in the candidate audit record. """ if not isinstance(fragment, dict): return fragment, [] normalized = deepcopy(fragment) fixes: list[dict[str, str]] = [] _lift_feature_local_sketches(normalized, fixes=fixes) feature_values: list[tuple[dict[str, Any], str]] = [] feature = normalized.get("feature") if isinstance(feature, dict): feature_values.append((feature, "feature")) features = normalized.get("features", normalized.get("add_features")) if isinstance(features, list): feature_values.extend( (item, f"features[{index}]") for index, item in enumerate(features) if isinstance(item, dict) ) for current_feature, location in feature_values: atomic_id = str(current_feature.get("atomic_id") or "") params = current_feature.get("params") if not isinstance(params, dict): continue if atomic_id in {"extrude_add_blind", "extrude_add_two_sided", "extrude_cut_blind"}: # CAD systems commonly call a blind extrusion's travel "depth". # CDSL calls the exact same signed scalar ``distance_mm``. _move_equivalent_field( params, source="depth_mm", target="distance_mm", location=f"{location}.params", fixes=fixes, ) if atomic_id in {"revolve_add", "revolve_cut"}: _move_equivalent_field( params, source="angle_degrees", target="angle_deg", location=f"{location}.params", fixes=fixes, ) axis = params.get("axis") if axis is None: axis = {} params["axis"] = axis if isinstance(axis, dict): _move_axis_component(params, axis, source="axis_origin_mm", target="origin_mm", location=f"{location}.params", fixes=fixes) _move_axis_component(params, axis, source="axis_point_mm", target="origin_mm", location=f"{location}.params", fixes=fixes) _move_axis_component(params, axis, source="axis_dir", target="direction", location=f"{location}.params", fixes=fixes) _move_axis_component(params, axis, source="axis_direction", target="direction", location=f"{location}.params", fixes=fixes) axis = params.get("axis") if isinstance(axis, dict): _normalize_axis_mapping(axis, location=f"{location}.params.axis", fixes=fixes) shared_axis = normalized.get("revolve_axis") if shared_axis is not None: if not isinstance(shared_axis, dict): raise AutonomousFragmentError("revolve_axis must be an explicit {origin_mm, direction} object") _normalize_axis_mapping(shared_axis, location="revolve_axis", fixes=fixes) revolved = [feature for feature, _ in feature_values if str(feature.get("atomic_id") or "").startswith("revolve_")] if not revolved: raise AutonomousFragmentError("revolve_axis is valid only in a fragment containing revolve_add or revolve_cut") for feature, location in feature_values: if not str(feature.get("atomic_id") or "").startswith("revolve_"): continue params = feature.get("params") if not isinstance(params, dict): continue axis = params.get("axis") if not isinstance(axis, dict) or not axis: params["axis"] = deepcopy(shared_axis) fixes.append({"path": f"{location}.params", "from": "revolve_axis", "to": "axis", "action": "copied_explicit_batch_axis"}) sketch_values: list[tuple[dict[str, Any], str]] = [] sketch = normalized.get("sketch") if isinstance(sketch, dict): sketch_values.append((sketch, "sketch")) sketches = normalized.get("sketches", normalized.get("add_sketches")) if isinstance(sketches, list): sketch_values.extend((item, f"sketches[{index}]") for index, item in enumerate(sketches) if isinstance(item, dict)) for current_sketch, location in sketch_values: profile = current_sketch.get("profile") if isinstance(profile, dict) and profile.get("type") == "polygon": _move_equivalent_field(profile, source="points", target="vertices", location=f"{location}.profile", fixes=fixes) return normalized, fixes def materialize_autonomous_fragment( base_cdsl: dict[str, Any] | None, fragment: dict[str, Any], *, engine: Any, selector_tokens: dict[str, dict[str, Any]], max_features: int, ) -> tuple[dict[str, Any], dict[str, Any]]: """Append authored geometry while assigning only server-owned metadata. Workplanes, profiles, feature parameters, directions and boolean meaning pass through exactly as the author supplied them. Local CDSL validation and full engine rebuild happen after this function, before a checkpoint exists. """ # Delayed import avoids engine_service -> selector_bindings -> this module # becoming an import cycle. from app.services.engine_service import feature_atomic_contract normalized_fragment, compatibility_fixes = normalize_autonomous_fragment(fragment) sketches, features = _fragment_lists(normalized_fragment) if len(features) > max_features: raise AutonomousFragmentError(f"A fragment may add at most {max_features} feature(s)") document = materialize_fragment(base_cdsl, {"add_sketches": [], "add_features": []}) geometry = document.get("geometry") if isinstance(document.get("geometry"), dict) else {} existing_sketches = geometry.get("sketches") if isinstance(geometry.get("sketches"), list) else [] existing_features = document.get("features") if isinstance(document.get("features"), list) else [] used_sketch_ids = {str(item.get("id") or "") for item in existing_sketches if isinstance(item, dict)} used_feature_ids = {str(item.get("id") or "") for item in existing_features if isinstance(item, dict)} last_feature_id = str(existing_features[-1].get("id") or "") if existing_features and isinstance(existing_features[-1], dict) else "" materialized_sketches: list[dict[str, Any]] = [] materialized_features: list[dict[str, Any]] = [] for index, source_feature in enumerate(features): forbidden = {"id", "depends_on", "sketch_id", "selectors"} & set(source_feature) if forbidden: raise AutonomousFragmentError("Feature identity, dependencies, sketch_id and raw selectors are server-owned: " + ", ".join(sorted(forbidden))) atomic_id = str(source_feature.get("atomic_id") or "") if not atomic_id: raise AutonomousFragmentError("Each fragment feature must declare a runtime atomic_id") contract = feature_atomic_contract(engine, atomic_id) params = source_feature.get("params") if not isinstance(params, dict): raise AutonomousFragmentError("Each fragment feature must contain a params object") if (atomic_id.startswith("hole_") or atomic_id == "hole_wizard") and "host_face" in params: raise AutonomousFragmentError( f"{atomic_id} host_face is server-owned: put exactly one face token in selector_tokens and use " "positions as [{\"mm\":[x_mm,y_mm,z_mm]}], not raw host_face or bare coordinate arrays" ) tokens = source_feature.pop("selector_tokens", []) if not isinstance(tokens, list) or not all(isinstance(token, str) for token in tokens) or len(set(tokens)) != len(tokens): raise AutonomousFragmentError("selector_tokens must be a unique array of opaque tokens") selected: list[dict[str, Any]] = [] for token in tokens: candidate = selector_tokens.get(token) if candidate is None: raise AutonomousFragmentError("TOPOLOGY_TOKEN_INVALID: selector token is not from the active snapshot") selected.append(deepcopy(candidate["selector"])) slot = contract.get("selector_slot") token_backed_param = atomic_id.startswith("hole_") or atomic_id == "hole_wizard" if not slot and tokens and not token_backed_param: raise AutonomousFragmentError(f"{atomic_id} does not accept selector tokens") if isinstance(slot, dict): minimum, maximum = int(slot.get("min_items") or 0), int(slot.get("max_items") or 0) if not minimum <= len(selected) <= maximum: raise AutonomousFragmentError(f"{atomic_id} requires {minimum}..{maximum} selector token(s)") elif atomic_id.startswith("hole_") or atomic_id == "hole_wizard": if len(selected) != 1 or str(selected[0].get("kind") or "") != "face": raise AutonomousFragmentError(f"{atomic_id} requires exactly one face selector token for its host face") elif atomic_id.startswith("revolve_"): axis = params.get("axis") if not isinstance(axis, dict) or "origin_mm" not in axis or "direction" not in axis: raise AutonomousFragmentError( f"{atomic_id} requires params.axis with explicit origin_mm and direction; " "the revolve axis is author-defined geometry, not a topology selector token" ) feature_id = _autonomous_id("feature", used_feature_ids) output = {key: deepcopy(value) for key, value in source_feature.items() if key != "selector_tokens"} output["id"] = feature_id output["depends_on"] = [last_feature_id] if last_feature_id else [] if contract["requires_sketch"]: if index >= len(sketches): raise AutonomousFragmentError(f"{atomic_id} requires one new sketch in the same fragment") sketch = sketches[index] if "id" in sketch or "attachment" in sketch or "profile_from" in sketch: raise AutonomousFragmentError("Sketch identity and topology attachment are server-owned") sketch_id = _autonomous_id("sketch", used_sketch_ids) sketch["id"] = sketch_id materialized_sketches.append(sketch) output["sketch_id"] = sketch_id elif index < len(sketches): raise AutonomousFragmentError(f"{atomic_id} does not accept a sketch") if atomic_id.startswith("pattern_") and "source_feature_ids" not in params: if not last_feature_id: raise AutonomousFragmentError(f"{atomic_id} needs a committed source feature") params["source_feature_ids"] = [last_feature_id] if isinstance(slot, dict) and slot.get("path") == "feature.selectors": output["selectors"] = selected elif isinstance(slot, dict) and slot.get("path") == "params.mirror_plane": output["params"]["mirror_plane"] = selected[0] output["selectors"] = [] elif isinstance(slot, dict) and slot.get("path") == "params.host_face": output["params"]["host_face"] = selected[0] output["selectors"] = [] elif atomic_id.startswith("hole_") or atomic_id == "hole_wizard": output["params"]["host_face"] = selected[0] output["selectors"] = [] else: output["selectors"] = [] materialized_features.append(output) last_feature_id = feature_id if len(sketches) != len(materialized_sketches): raise AutonomousFragmentError("Each sketch must be consumed by a feature that requires a sketch") document = materialize_fragment(document, {"add_sketches": materialized_sketches, "add_features": materialized_features}) return document, { "schema_version": "cad.autonomous-fragment.v1", "source_fragment": deepcopy(fragment), "normalized_fragment": deepcopy(normalized_fragment) if compatibility_fixes else None, "compatibility_fixes": compatibility_fixes, "assigned_sketch_ids": [item["id"] for item in materialized_sketches], "assigned_feature_ids": [item["id"] for item in materialized_features], "selector_candidate_ids": [token for feature in features for token in feature.get("selector_tokens", [])], }