change

backend/engine/cdsl_engine/semantic_validation.py

@@ -0,0 +1,103 @@
+"""Validation for the complete CDSL v1.1 semantic contract.
+
+The current runtime accepts only a subset of this contract.  Keeping this
+validator separate lets import tooling preserve a SolidWorks feature history
+without claiming that every feature can already be rebuilt locally.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+from jsonschema import Draft202012Validator
+
+
+_ID = re.compile(r"^[A-Za-z0-9_-]{1,80}$")
+
+
+@lru_cache(maxsize=1)
+def _schema() -> dict[str, Any]:
+    path = Path(__file__).with_name("cdsl_schema.json")
+    schema = json.loads(path.read_text(encoding="utf-8"))
+    Draft202012Validator.check_schema(schema)
+    return schema
+
+
+@lru_cache(maxsize=1)
+def _validator() -> Draft202012Validator:
+    return Draft202012Validator(_schema())
+
+
+def _schema_error(document: dict[str, Any]) -> str | None:
+    validator = _validator()
+    errors = sorted(validator.iter_errors(document), key=lambda error: (list(error.absolute_path), error.message))
+    if not errors:
+        return None
+    error = errors[0]
+    location = "$" + "".join(f"[{item}]" if isinstance(item, int) else f".{item}" for item in error.absolute_path)
+    return f"CDSL schema violation at {location}: {error.message}"
+
+
+def validate_semantic_cdsl(cdsl: dict[str, Any]) -> dict[str, Any]:
+    """Validate a CDSL document without invoking the rebuild compiler.
+
+    The return value is intentionally serializable so the batch converter can
+    write it unchanged into a per-model diagnostic file.
+    """
+    if not isinstance(cdsl, dict):
+        raise ValueError("CDSL must be a JSON object")
+    if cdsl.get("schema") != "cad.cdsl.llm.v1":
+        raise ValueError("Unsupported CDSL schema")
+    schema_error = _schema_error(cdsl)
+    if schema_error:
+        raise ValueError(schema_error)
+
+    version = str(cdsl.get("schema_version") or "1.0.0")
+    if not re.fullmatch(r"1\.[0-9]+\.[0-9]+", version):
+        raise ValueError("schema_version must be a 1.x.y version")
+    version_numbers = tuple(int(component) for component in version.split("."))
+    if version_numbers >= (1, 1, 0) and cdsl.get("meta", {}).get("unit") != "mm":
+        raise ValueError("CDSL v1.1 requires meta.unit = 'mm'")
+
+    sketches = (cdsl.get("geometry") or {}).get("sketches") or []
+    sketch_ids = {str(sketch.get("id") or "") for sketch in sketches}
+    if len(sketch_ids) != len(sketches) or not all(_ID.fullmatch(item) for item in sketch_ids):
+        raise ValueError("Sketch ids must be unique valid CDSL identifiers")
+
+    feature_ids: set[str] = set()
+    deferred: list[str] = []
+    unresolved: list[dict[str, Any]] = []
+    for feature in cdsl.get("features") or []:
+        fid = str(feature.get("id") or "")
+        if not _ID.fullmatch(fid) or fid in feature_ids:
+            raise ValueError("Feature ids must be unique valid CDSL identifiers")
+        for dependency in feature.get("depends_on") or []:
+            if dependency not in feature_ids:
+                raise ValueError(f"Feature {fid} has a forward or missing dependency: {dependency}")
+        sketch_id = feature.get("sketch_id")
+        if sketch_id is not None and str(sketch_id) not in sketch_ids:
+            raise ValueError(f"Feature {fid} refers to a missing sketch: {sketch_id}")
+        if version_numbers >= (1, 1, 0) and feature.get("execution_status") not in {"supported", "deferred"}:
+            raise ValueError(f"Feature {fid} must declare execution_status")
+        if feature.get("execution_status") == "deferred":
+            deferred.append(fid)
+        for index, selector in enumerate(feature.get("selectors") or []):
+            owner = selector.get("owner_feature_id")
+            if owner is not None and owner not in feature_ids:
+                raise ValueError(f"Feature {fid} selector {index} has a forward or missing owner_feature_id")
+        if feature.get("unresolved"):
+            unresolved.append({"feature_id": fid, "reasons": list(feature["unresolved"])})
+        feature_ids.add(fid)
+
+    return {
+        "schema_version": version,
+        "feature_count": len(feature_ids),
+        "sketch_count": len(sketches),
+        "deferred_feature_ids": deferred,
+        "unresolved": unresolved,
+        "future_rebuild_ready": not unresolved,
+    }