app/services/composition_compiler.py

from __future__ import annotations

import ast
import math
import re
import time
from dataclasses import dataclass

from app.schemas.composition import (
    AnnotationPanelSpec,
    Bounds3D,
    ChartEncoding,
    ChartPanelSpec,
    ChartSpec,
    CompositionAssertion,
    CompositionPayload,
    CompositionSpec,
    DatasetSpec,
    DerivedValueSpec,
    DiagramPanelSpec,
    EquationPanelSpec,
    ExpressionInstruction,
    ExpressionProgram,
    GeometryPresentation,
    ImplicitGeometrySpec,
    InteractionSpec,
    MathematicalProperty,
    ParameterConstraints,
    ParameterSpec,
    ParametricGeometrySpec,
    ProvenanceSpec,
    RegisteredGeometrySpec,
    RequirementCoverage,
    Resolution3D,
    SemanticDiagramEdge,
    SemanticDiagramNode,
    SemanticDiagramSpec,
    ThreeScenePanelSpec,
    VisualizationIntentRecord,
    VectorFieldGeometrySpec,
)
from app.schemas.visual_lesson import ExtractedChartDataset


class CompositionCompilationError(ValueError):
    code = "composition_compilation_failed"


_FUNCTIONS = {"sin", "cos", "tan", "sqrt", "abs", "exp", "log", "min", "max"}
_VARIABLES = {"x", "y", "z", "u", "v", "pi", "a", "b", "c", "e", "iso_value"}
_BINOPS = {
    ast.Add: "add", ast.Sub: "subtract", ast.Mult: "multiply", ast.Div: "divide", ast.Pow: "power",
}


class SafeExpressionParser:
    """Turns user-supplied mathematics into nonrecursive, executable-free instructions."""

    @staticmethod
    def _normalise(source: str) -> str:
        source = source.strip().replace("−", "-").replace("×", "*").replace("÷", "/")
        source = source.replace("²", "^2").replace("³", "^3").replace("π", "pi").replace("^", "**")
        source = re.sub(r"(?<=\d)(?=[a-zA-Z(])|(?<=[xyzuv)])(?=\()", "*", source)
        return source

    def parse_equation(self, source: str) -> ExpressionProgram:
        normalised = self._normalise(source)
        if "=" in normalised:
            left, right = normalised.split("=", 1)
            normalised = f"({left})-({right})"
        if len(normalised) > 500:
            raise CompositionCompilationError("The supplied expression is too long")
        try:
            tree = ast.parse(normalised, mode="eval")
        except SyntaxError as exc:
            raise CompositionCompilationError("The supplied equation could not be parsed") from exc
        instructions: list[ExpressionInstruction] = []

        def emit(node: ast.AST) -> str:
            instruction_id = f"e{len(instructions)}"
            if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
                value = float(node.value)
                if not math.isfinite(value) or abs(value) > 1e12:
                    raise CompositionCompilationError("Expression constants must be finite and bounded")
                instructions.append(ExpressionInstruction(instruction_id=instruction_id, op="constant", value=value))
                return instruction_id
            if isinstance(node, ast.Name) and node.id in _VARIABLES:
                instructions.append(ExpressionInstruction(instruction_id=instruction_id, op="variable", name=node.id))
                return instruction_id
            if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
                arg = emit(node.operand)
                instruction_id = f"e{len(instructions)}"
                instructions.append(ExpressionInstruction(instruction_id=instruction_id, op="negate", args=[arg]))
                return instruction_id
            if isinstance(node, ast.BinOp) and type(node.op) in _BINOPS:
                left, right = emit(node.left), emit(node.right)
                if isinstance(node.op, ast.Pow) and isinstance(node.right, ast.Constant) and abs(float(node.right.value)) > 12:
                    raise CompositionCompilationError("Expression exponents are limited to an absolute value of 12")
                instruction_id = f"e{len(instructions)}"
                instructions.append(ExpressionInstruction(instruction_id=instruction_id, op=_BINOPS[type(node.op)], args=[left, right]))
                return instruction_id
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in _FUNCTIONS:
                args = [emit(item) for item in node.args]
                if node.func.id not in {"min", "max"} and len(args) != 1:
                    raise CompositionCompilationError(f"{node.func.id} expects one argument")
                if node.func.id in {"min", "max"} and len(args) < 2:
                    raise CompositionCompilationError(f"{node.func.id} expects at least two arguments")
                instruction_id = f"e{len(instructions)}"
                instructions.append(ExpressionInstruction(instruction_id=instruction_id, op=node.func.id, args=args))
                return instruction_id
            raise CompositionCompilationError("The expression contains an unsupported construct")

        result_id = emit(tree.body)
        if len(instructions) > 128:
            raise CompositionCompilationError("The expression exceeds the 128-operation limit")
        return ExpressionProgram(instructions=instructions, result_id=result_id)


class CompositionValidator:
    MAX_ROWS = 20_000
    MAX_GRID_SAMPLES = 2_100_000

    def validate(self, spec: CompositionSpec) -> None:
        parameter_ids = {item.parameter_id for item in spec.parameters}
        if len(parameter_ids) != len(spec.parameters):
            raise CompositionCompilationError("Composition parameter IDs must be unique")
        panel_ids = {panel.panel_id for panel in spec.panels}
        if len(panel_ids) != len(spec.panels) or not panel_ids:
            raise CompositionCompilationError("Composition panel IDs must be unique and nonempty")
        dataset_ids = {dataset.dataset_id for dataset in spec.datasets}
        for dataset in spec.datasets:
            if len(dataset.rows) > self.MAX_ROWS:
                raise CompositionCompilationError("Chart dataset exceeds the interactive row budget")
            if any(set(row) - set(dataset.columns) for row in dataset.rows):
                raise CompositionCompilationError("Chart rows contain undeclared columns")
        for panel in spec.panels:
            if set(panel.bindings) - (parameter_ids | {item.value_id for item in spec.derived_values}):
                raise CompositionCompilationError(f"Panel {panel.panel_id} references an unknown binding")
            if isinstance(panel, ChartPanelSpec) and panel.chart.dataset_id not in dataset_ids:
                raise CompositionCompilationError("Chart panel references an unknown dataset")
            if isinstance(panel, DiagramPanelSpec) and panel.diagram_type == "semantic":
                diagram = panel.semantic
                if diagram is None:
                    raise CompositionCompilationError("Semantic diagram panels require a semantic diagram specification")
                if not 1 <= len(diagram.nodes) <= 80 or len(diagram.edges) > 160:
                    raise CompositionCompilationError("Semantic diagram exceeds the interactive node or edge budget")
                node_ids = {node.node_id for node in diagram.nodes}
                if len(node_ids) != len(diagram.nodes):
                    raise CompositionCompilationError("Semantic diagram node IDs must be unique")
                if any(edge.source not in node_ids or edge.target not in node_ids for edge in diagram.edges):
                    raise CompositionCompilationError("Semantic diagram edge references an unknown node")
            if isinstance(panel, ChartPanelSpec):
                dataset = next(item for item in spec.datasets if item.dataset_id == panel.chart.dataset_id)
                available = set(dataset.columns)
                for transform in panel.chart.transforms:
                    if transform.input_field and transform.input_field not in available:
                        raise CompositionCompilationError("Chart transform references an unknown input field")
                    if transform.kind == "bin" and not 2 <= transform.bins <= 200:
                        raise CompositionCompilationError("Chart bin counts must stay between 2 and 200")
                    if transform.output_field:
                        available.add(transform.output_field)
            if isinstance(panel, ThreeScenePanelSpec) and isinstance(panel.geometry, ImplicitGeometrySpec):
                r = panel.geometry.resolution
                if min(r.preview, r.refined, r.high_quality) < 8 or max(r.preview, r.refined, r.high_quality) > 128:
                    raise CompositionCompilationError("Implicit resolution must stay between 8 and 128")
                if r.refined ** 3 > self.MAX_GRID_SAMPLES:
                    raise CompositionCompilationError("Implicit resolution exceeds the scalar-field budget")
                self.validate_program(panel.geometry.expression)
            if isinstance(panel, ThreeScenePanelSpec) and isinstance(panel.geometry, ParametricGeometrySpec):
                if panel.geometry.u_segments < 4 or panel.geometry.v_segments < 4 or panel.geometry.u_segments * panel.geometry.v_segments > 100_000:
                    raise CompositionCompilationError("Parametric surface resolution exceeds its budget")
                self.validate_program(panel.geometry.x_program)
                self.validate_program(panel.geometry.y_program)
                self.validate_program(panel.geometry.z_program)
            if isinstance(panel, ThreeScenePanelSpec) and isinstance(panel.geometry, VectorFieldGeometrySpec):
                if not 2 <= panel.geometry.samples_per_axis <= 32:
                    raise CompositionCompilationError("Vector-field sampling must stay between 2 and 32 per axis")
                self.validate_program(panel.geometry.x_program)
                self.validate_program(panel.geometry.y_program)
                self.validate_program(panel.geometry.z_program)
        for value in spec.derived_values:
            self.validate_program(value.program)
        for interaction in spec.interactions:
            if interaction.state_id not in parameter_ids and interaction.kind == "parameter":
                raise CompositionCompilationError("Parameter interaction references an unknown parameter")
            if set(interaction.target_panel_ids) - panel_ids:
                raise CompositionCompilationError("Interaction references an unknown panel")
        if spec.request_intent:
            covered = {item.requirement for item in spec.requirement_coverage if item.status == "satisfied"}
            missing = set(spec.request_intent.hard_requirements) - covered
            if missing:
                raise CompositionCompilationError(
                    "Composition does not satisfy required visualization commands: " + ", ".join(sorted(missing))
                )
        if any(item.status != "satisfied" for item in spec.requirement_coverage):
            raise CompositionCompilationError("A required visualization command is unsupported or conflicting")

    @staticmethod
    def validate_program(program: ExpressionProgram) -> None:
        seen: set[str] = set()
        for instruction in program.instructions:
            if instruction.instruction_id in seen:
                raise CompositionCompilationError("Expression instruction IDs must be unique")
            if any(arg not in seen for arg in instruction.args):
                raise CompositionCompilationError("Expression instructions must form a forward-only DAG")
            if instruction.op == "constant" and (instruction.value is None or not math.isfinite(instruction.value)):
                raise CompositionCompilationError("Expression constants must be finite")
            if instruction.op == "variable" and instruction.name not in _VARIABLES:
                raise CompositionCompilationError("Expression variable is not approved")
            seen.add(instruction.instruction_id)
        if program.result_id not in seen:
            raise CompositionCompilationError("Expression result does not exist")

    def validate_checkpoint(self, spec: CompositionSpec, values: dict[str, object]) -> None:
        parameters = {item.parameter_id: item for item in spec.parameters}
        if set(values) - set(parameters):
            raise CompositionCompilationError("Checkpoint contains an unknown parameter")
        resolved = {item.parameter_id: item.initial_value for item in spec.parameters}
        resolved.update(values)
        for parameter_id, value in resolved.items():
            parameter = parameters[parameter_id]
            if parameter.value_type == "boolean" and not isinstance(value, bool):
                raise CompositionCompilationError(f"{parameter.label} must be true or false")
            if parameter.value_type in {"scalar", "integer"}:
                if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)):
                    raise CompositionCompilationError(f"{parameter.label} must be finite")
                number = float(value)
                c = parameter.constraints
                if c.minimum is not None and (number < c.minimum or (c.exclusive_minimum and number == c.minimum)):
                    raise CompositionCompilationError(f"{parameter.label} is below its allowed range")
                if c.maximum is not None and (number > c.maximum or (c.exclusive_maximum and number == c.maximum)):
                    raise CompositionCompilationError(f"{parameter.label} is above its allowed range")
            if parameter.value_type == "enum" and str(value) not in parameter.constraints.allowed_values:
                raise CompositionCompilationError(f"{parameter.label} is not an allowed option")
        operator = next((p.geometry for p in spec.panels if isinstance(p, ThreeScenePanelSpec) and isinstance(p.geometry, RegisteredGeometrySpec)), None)
        if operator and operator.operator == "prolate_spheroid":
            a = float(resolved[operator.parameter_bindings["majorSemiAxis"]])
            b = float(resolved[operator.parameter_bindings["equatorialSemiAxis"]])
            if not a > b > 0:
                raise CompositionCompilationError("A prolate spheroid requires a > b > 0")


@dataclass(frozen=True)
class ChartRendererDefinition:
    renderer: str
    families: frozenset[str]


CHART_RENDERERS = (
    ChartRendererDefinition("plotly", frozenset({"line", "scatter", "area", "bar", "histogram", "box", "violin", "heatmap", "contour", "error-band"})),
    ChartRendererDefinition("d3", frozenset({"force-graph", "tree", "sankey", "hierarchy", "chord", "treemap", "packed-circles"})),
    ChartRendererDefinition("svg", frozenset({"heatmap"})),
)


def chart_renderer_for(spec: ChartSpec) -> str:
    candidates = [item.renderer for item in CHART_RENDERERS if spec.family in item.families]
    if spec.renderer_preference != "automatic" and spec.renderer_preference in candidates:
        return spec.renderer_preference
    if not candidates:
        raise CompositionCompilationError(f"No verified renderer supports {spec.family}")
    return candidates[0]


class CompositionCompiler:
    GEOMETRY_TERMS = {
        "prolate_spheroid": ("prolate spheroid", "prolate ellipsoid"),
        "mobius_strip": ("möbius", "mobius"),
        "torus": ("torus", "toroid", "doughnut surface"),
        "ellipsoid": ("ellipsoid", "spheroid"),
        "sphere": ("sphere", "spherical surface"),
    }

    def __init__(self) -> None:
        self.parser = SafeExpressionParser()
        self.validator = CompositionValidator()

    @classmethod
    def supports(cls, prompt: str) -> bool:
        lowered = prompt.lower()
        if any(term in lowered for terms in cls.GEOMETRY_TERMS.values() for term in terms):
            return True
        if any(term in lowered for term in ("line chart", "scatter plot", "bar chart", "heatmap", "force graph", "force-directed", "parabola", "quadratic curve", "quadratic function")):
            return True
        if all(re.search(rf"\b{name}\s*=", lowered) for name in ("x", "y", "z")) and any(variable in lowered for variable in ("u", "v")):
            return True
        return "=" in prompt and any(variable in lowered for variable in ("x", "y", "z"))

    @staticmethod
    def _common_parameters(prompt: str = "") -> list[ParameterSpec]:
        lowered = prompt.lower()
        opacity_match = re.search(r"\bopacity\s*(?:=|of|to)?\s*(0(?:\.\d+)?|1(?:\.0+)?)", lowered)
        opacity = float(opacity_match.group(1)) if opacity_match else 0.88
        return [
            ParameterSpec(parameter_id="opacity", label="Opacity", value_type="scalar", initial_value=opacity, constraints=ParameterConstraints(minimum=0.1, maximum=1), recompute="presentation-only"),
            ParameterSpec(parameter_id="wireframe", label="Wireframe", value_type="boolean", initial_value="wireframe" in lowered and not any(term in lowered for term in ("no wireframe", "without wireframe", "wireframe off")), recompute="presentation-only"),
            ParameterSpec(parameter_id="show_axes", label="Axes", value_type="boolean", initial_value=not any(term in lowered for term in ("hide axes", "without axes", "no axes", "axes off")), recompute="presentation-only"),
            ParameterSpec(parameter_id="spin", label="Spin", value_type="boolean", initial_value=any(term in lowered for term in ("spin", "rotate automatically", "auto-rotate", "auto rotate")), recompute="presentation-only"),
        ]

    def build(self, project_id: str, prompt: str, extracted: ExtractedChartDataset | None = None) -> CompositionPayload:
        lowered = prompt.lower()
        if extracted is not None or any(term in lowered for term in ("line chart", "scatter plot", "bar chart", "heatmap", "force graph", "force-directed", "parabola", "quadratic curve", "quadratic function")):
            spec = self._chart(project_id, prompt, extracted=extracted)
        elif all(re.search(rf"\b{name}\s*=", lowered) for name in ("x", "y", "z")) and any(variable in lowered for variable in ("u", "v")):
            spec = self._parametric(project_id, prompt)
        elif "=" in prompt and any(variable in lowered for variable in ("x", "y", "z")):
            spec = self._implicit(project_id, prompt)
        else:
            operator = next((name for name, terms in self.GEOMETRY_TERMS.items() if any(term in lowered for term in terms)), "")
            if not operator:
                raise CompositionCompilationError("The requested composition has no registered deterministic operator")
            spec = self._registered(project_id, prompt, operator)
        self.validator.validate(spec)
        return CompositionPayload(spec=spec, warnings=[])

    def build_planned(self, project_id: str, prompt: str, kind: str, target: str, extracted: ExtractedChartDataset | None = None) -> CompositionPayload:
        if kind == "registered_geometry" and target in self.GEOMETRY_TERMS:
            spec = self._registered(project_id, prompt, target)
        elif kind == "chart" and target in {"line", "scatter", "bar", "heatmap", "force-graph"}:
            spec = self._chart(project_id, prompt, target, extracted=extracted)
        else:
            raise CompositionCompilationError("No registered composition operator can answer this request")
        self.validator.validate(spec)
        return CompositionPayload(spec=spec, warnings=[])

    _REQUIREMENT_STOPWORDS = {
        "the", "a", "an", "and", "or", "with", "for", "of", "to", "in", "on", "at", "by",
        "show", "add", "include", "make", "generate", "create", "render", "featuring",
        "should", "must", "that", "this", "each", "every", "have", "has", "can", "using",
    }

    @classmethod
    def _significant_words(cls, text: str) -> set[str]:
        words = re.findall(r"[a-z][a-z0-9]{2,}", text.lower())
        return {word for word in words if word not in cls._REQUIREMENT_STOPWORDS}

    @classmethod
    def _facet_text(cls, facet) -> str:
        parts = [facet.title, facet.purpose]
        parts.extend(node.label for node in facet.nodes)
        parts.extend(edge.label for edge in facet.edges if edge.label)
        return " ".join(parts)

    def build_semantic(self, project_id: str, prompt: str, intent, plan) -> CompositionPayload:
        """Compile an agent-selected semantic plan into app-owned diagram panels."""
        panels: list[DiagramPanelSpec] = []
        for order, facet in enumerate(plan.facets):
            nodes = [SemanticDiagramNode(**node.model_dump()) for node in facet.nodes]
            edges = [SemanticDiagramEdge(**edge.model_dump()) for edge in facet.edges]
            semantic = SemanticDiagramSpec(
                diagram_id=facet.facet_id,
                title=facet.title,
                layout=facet.layout,
                nodes=nodes,
                edges=edges,
            )
            panels.append(DiagramPanelSpec(
                panel_id=facet.facet_id,
                diagram_type="semantic",
                title=facet.title,
                semantic=semantic,
                order=order,
            ))
        # A semantic panel has no authority to prove raw natural-language commands such
        # as "3D", "animation", or "plot". Those modes are resolved before this
        # compiler runs; keeping phrase-overlap coverage here created false acceptance
        # and false rejection alike.
        coverage: list[RequirementCoverage] = []
        spec = CompositionSpec(
            project_id=project_id,
            prompt=prompt,
            title=plan.title,
            answer_markdown=plan.answer_markdown,
            parameters=[],
            panels=panels,
            interactions=[],
            assumptions=["Diagram geometry is schematic; nodes and edges represent the requested mechanism rather than measured spatial coordinates."],
            provenance=ProvenanceSpec(interpretation=plan.interpretation, research_used=False),
            request_intent=VisualizationIntentRecord(
                specificity=intent.specificity,
                interpretation=intent.interpretation,
                requested_views=intent.requested_views,
                hard_requirements=[],
                requested_interactions=intent.requested_interactions,
            ),
            requirement_coverage=coverage,
            created_at=time.time(),
        )
        self.validator.validate(spec)
        return CompositionPayload(spec=spec, warnings=[])

    def _registered(self, project_id: str, prompt: str, operator: str) -> CompositionSpec:
        parameters = self._common_parameters(prompt)
        bindings: dict[str, str]
        derived: list[DerivedValueSpec] = []
        panels: list = []
        properties: list[MathematicalProperty] = []
        equations: list[str] = []
        title = operator.replace("_", " ").title()
        assumptions = ["Dimensions are illustrative unless the prompt supplies explicit values."]
        if operator == "prolate_spheroid":
            a_match = re.search(r"\ba\s*=\s*(\d+(?:\.\d+)?)", prompt, flags=re.I)
            b_match = re.search(r"\bb\s*=\s*(\d+(?:\.\d+)?)", prompt, flags=re.I)
            axes_match = re.search(r"\b(?:semi[- ]?axes|axes)\s+(\d+(?:\.\d+)?)\s+(?:and|by|,|×|x)\s*(\d+(?:\.\d+)?)", prompt, flags=re.I)
            a_value = float(a_match.group(1)) if a_match else float(axes_match.group(1)) if axes_match else 2.0
            b_value = float(b_match.group(1)) if b_match else float(axes_match.group(2)) if axes_match else 1.0
            if a_value <= b_value:
                raise CompositionCompilationError("A prolate spheroid requires the major semi-axis a to be greater than b")
            parameters = [
                ParameterSpec(parameter_id="a", label="Major semi-axis a", value_type="scalar", initial_value=a_value, constraints=ParameterConstraints(minimum=0.05, maximum=max(10, a_value), exclusive_minimum=True), recompute="geometry-transform", persistence="lesson"),
                ParameterSpec(parameter_id="b", label="Equatorial semi-axis b", value_type="scalar", initial_value=b_value, constraints=ParameterConstraints(minimum=0.05, maximum=max(10, b_value), exclusive_minimum=True), recompute="geometry-transform", persistence="lesson"),
                *parameters,
            ]
            bindings = {"majorSemiAxis": "a", "equatorialSemiAxis": "b"}
            derived = [
                DerivedValueSpec(value_id="c", label="Focal distance", program=self.parser.parse_equation("sqrt(a^2-b^2)")),
                DerivedValueSpec(value_id="e", label="Eccentricity", program=self.parser.parse_equation("sqrt(a^2-b^2)/a")),
                DerivedValueSpec(value_id="volume", label="Volume", program=self.parser.parse_equation("4*pi*a*b^2/3")),
            ]
            equations = ["x^2/b^2 + y^2/b^2 + z^2/a^2 = 1", "c = sqrt(a^2-b^2)", "e = c/a", "V = 4*pi*a*b^2/3"]
            properties = [
                MathematicalProperty(name="surface type", value="closed quadric", status="registered", method="Canonical prolate-spheroid operator"),
                MathematicalProperty(name="rotational symmetry", value=True, status="registered", method="Revolution about the major axis"),
            ]
            panels = [
                ThreeScenePanelSpec(panel_id="shape", geometry=RegisteredGeometrySpec(operator="prolate_spheroid", parameter_bindings=bindings), bindings=["a", "b", "c", "e", "opacity", "wireframe", "show_axes", "spin"], order=0),
                DiagramPanelSpec(panel_id="cross-section", diagram_type="ellipse_cross_section", bindings=["a", "b", "c"], order=1),
                EquationPanelSpec(panel_id="equations", equations=equations, bindings=["a", "b", "c", "e", "volume"], order=2),
            ]
        else:
            parameters.insert(0, ParameterSpec(parameter_id="scale", label="Scale", value_type="scalar", initial_value=1.5, constraints=ParameterConstraints(minimum=0.1, maximum=8, exclusive_minimum=True), recompute="geometry-transform", persistence="lesson"))
            bindings = {"scale": "scale"}
            if operator == "ellipsoid":
                parameters[0:1] = [
                    ParameterSpec(parameter_id="a", label="Axis a", value_type="scalar", initial_value=2.0, constraints=ParameterConstraints(minimum=0.1, maximum=8), recompute="geometry-transform", persistence="lesson"),
                    ParameterSpec(parameter_id="b", label="Axis b", value_type="scalar", initial_value=1.4, constraints=ParameterConstraints(minimum=0.1, maximum=8), recompute="geometry-transform", persistence="lesson"),
                    ParameterSpec(parameter_id="c", label="Axis c", value_type="scalar", initial_value=1.0, constraints=ParameterConstraints(minimum=0.1, maximum=8), recompute="geometry-transform", persistence="lesson"),
                ]
                bindings = {"xSemiAxis": "a", "ySemiAxis": "b", "zSemiAxis": "c"}
            panels = [ThreeScenePanelSpec(panel_id="shape", geometry=RegisteredGeometrySpec(operator=operator, parameter_bindings=bindings), bindings=[*bindings.values(), "opacity", "wireframe", "show_axes", "spin"], order=0)]
            properties = [MathematicalProperty(name="operator", value=operator, status="registered", method="App-owned geometry registry")]
        panel_ids = [panel.panel_id for panel in panels]
        interactions = [InteractionSpec(interaction_id=f"change-{item.parameter_id}", kind="parameter", state_id=item.parameter_id, target_panel_ids=panel_ids) for item in parameters]
        return CompositionSpec(
            project_id=project_id, prompt=prompt, title=title,
            answer_markdown=f"I interpreted your request as an interactive **{title.lower()}**. Its controls update every bound view from the same verified parameter state.",
            parameters=parameters, derived_values=derived, panels=panels, interactions=interactions,
            assumptions=assumptions, provenance=ProvenanceSpec(interpretation=f"Registered analytic operator: {operator}", operator_sources=[operator]),
            assertions=[CompositionAssertion(assertion_id="finite-derived", kind="finite", target_id="derived")],
            mathematical_properties=properties, created_at=time.time(),
        )

    def _implicit(self, project_id: str, prompt: str) -> CompositionSpec:
        match = re.search(r"([^\n.;]+=[^\n.;]+)", prompt)
        equation = (match.group(1) if match else prompt).strip()
        equation = re.sub(r"^(?:visuali[sz]e|plot|show)\s+", "", equation, flags=re.I)
        program = self.parser.parse_equation(equation)
        parameters = [
            ParameterSpec(parameter_id="iso_value", label="Iso-value", value_type="scalar", initial_value=0.0, constraints=ParameterConstraints(minimum=-100, maximum=100), recompute="remesh", persistence="lesson"),
            ParameterSpec(parameter_id="high_quality", label="96³ quality", value_type="boolean", initial_value=False, recompute="field-resample", persistence="session"),
            *self._common_parameters(prompt),
        ]
        geometry = ImplicitGeometrySpec(
            expression=program, bounds=Bounds3D(x=(-2, 2), y=(-2, 2), z=(-2, 2)), resolution=Resolution3D(),
            presentation=GeometryPresentation(show_bounding_box=True),
        )
        panel = ThreeScenePanelSpec(panel_id="implicit-surface", geometry=geometry, bindings=[item.parameter_id for item in parameters], order=0)
        return CompositionSpec(
            project_id=project_id, prompt=prompt, title="Implicit surface", answer_markdown=f"I rendered the zero level set of `{equation}` inside the finite domain x, y, z ∈ [-2, 2]. The domain and sampling resolution affect the visible approximation.",
            parameters=parameters, panels=[panel], interactions=[InteractionSpec(interaction_id=f"change-{item.parameter_id}", kind="parameter", state_id=item.parameter_id, target_panel_ids=[panel.panel_id]) for item in parameters],
            assumptions=["The finite plotting domain is x, y, z ∈ [-2, 2].", "The surface is a marching-cubes approximation, not exact symbolic geometry."],
            provenance=ProvenanceSpec(interpretation="User-supplied implicit equation", research_used=False),
            assertions=[CompositionAssertion(assertion_id="mesh-budget", kind="mesh_budget", target_id="implicit-surface")],
            mathematical_properties=[MathematicalProperty(name="topology", value=None, status="unknown", caveat="Topology is not inferred from the intended equation or a finite-resolution mesh.")], created_at=time.time(),
        )

    def _parametric(self, project_id: str, prompt: str) -> CompositionSpec:
        expressions: dict[str, str] = {}
        for name in ("x", "y", "z"):
            match = re.search(rf"\b{name}\s*=\s*([^;\n]+)", prompt, flags=re.I)
            if not match:
                raise CompositionCompilationError("Parametric surfaces require explicit x(u,v), y(u,v), and z(u,v) expressions separated by semicolons")
            expressions[name] = match.group(1).strip()
        geometry = ParametricGeometrySpec(
            x_program=self.parser.parse_equation(expressions["x"]),
            y_program=self.parser.parse_equation(expressions["y"]),
            z_program=self.parser.parse_equation(expressions["z"]),
            u_domain=(0, 2 * math.pi), v_domain=(0, math.pi), u_segments=80, v_segments=40,
        )
        parameters = self._common_parameters(prompt)
        panel = ThreeScenePanelSpec(panel_id="parametric-surface", geometry=geometry, bindings=[item.parameter_id for item in parameters])
        return CompositionSpec(
            project_id=project_id, prompt=prompt, title="Parametric surface",
            answer_markdown="I rendered the supplied parametric surface over the disclosed finite u/v domain. The mesh resolution controls its numerical approximation.",
            parameters=parameters, panels=[panel], interactions=[InteractionSpec(interaction_id=f"change-{item.parameter_id}", kind="parameter", state_id=item.parameter_id, target_panel_ids=[panel.panel_id]) for item in parameters],
            assumptions=["The parameter domain is u ∈ [0, 2π], v ∈ [0, π] unless a registered operator provides another domain."],
            provenance=ProvenanceSpec(interpretation="User-supplied parametric equations"),
            mathematical_properties=[MathematicalProperty(name="topology", value=None, status="unknown", caveat="No topological property is inferred from sampled parametric geometry.")], created_at=time.time(),
        )
    @staticmethod
    def _is_number(text: str) -> bool:
        try:
            float(text)
            return True
        except ValueError:
            return False

    def _chart(self, project_id: str, prompt: str, family_override: str = "", extracted: ExtractedChartDataset | None = None) -> CompositionSpec:
        lowered = prompt.lower()
        keyword_family = "force-graph" if "force" in lowered else "heatmap" if "heatmap" in lowered else "scatter" if "scatter" in lowered else "bar" if "bar" in lowered else "line"
        has_grounded_data = extracted is not None and extracted.has_data and bool(extracted.rows)
        if family_override:
            family = family_override
        elif keyword_family == "force-graph":
            # An explicit network-graph request never fits label/value rows -> keep the deterministic path.
            family = "force-graph"
        elif has_grounded_data:
            # Trust the data-informed choice over a keyword guess that just defaults to "line".
            family = extracted.chart_family
        else:
            family = keyword_family
        is_parabola = any(term in lowered for term in ("parabola", "quadratic curve", "quadratic function"))
        grounded_rows = extracted.rows if (has_grounded_data and family != "force-graph") else []
        x_label, y_label = "x", "y = x²" if is_parabola else "y"
        if grounded_rows:
            illustrative = False
            x_is_numeric = all(self._is_number(row.label) for row in grounded_rows)
            rows = [{"x": float(row.label) if x_is_numeric else row.label, "y": row.value} for row in grounded_rows]
            columns = ["x", "y"]
            encodings = [
                ChartEncoding(channel="x", field="x", value_type="quantitative" if x_is_numeric else "nominal"),
                ChartEncoding(channel="y", field="y"),
            ]
            x_label = extracted.x_label or x_label
            y_label = extracted.y_label or y_label
        else:
            pairs = [(float(a), float(b)) for a, b in re.findall(r"\(?\s*(-?\d+(?:\.\d+)?)\s*[, :]\s*(-?\d+(?:\.\d+)?)\s*\)?", prompt)][:200]
            illustrative = not pairs
            pairs = pairs or ([(float(x), float(x * x)) for x in range(-5, 6)] if is_parabola else [(float(x), float(x * x)) for x in range(6)])
            if family == "force-graph":
                named_edges = re.findall(r"\b([A-Za-z][\w-]*)\s*(?:->|--|—|–)\s*([A-Za-z][\w-]*)\b", prompt)
                rows = ([{"source": source, "target": target, "weight": 1.0} for source, target in named_edges]
                        if named_edges else [{"source": f"N{int(x)}", "target": f"N{int(y)}", "weight": 1.0} for x, y in pairs])
                illustrative = not named_edges and illustrative
                columns = ["source", "target", "weight"]
                encodings = [ChartEncoding(channel="source", field="source", value_type="nominal"), ChartEncoding(channel="target", field="target", value_type="nominal"), ChartEncoding(channel="weight", field="weight")]
            else:
                rows = [{"x": x, "y": y} for x, y in pairs]
                columns = ["x", "y"]
                encodings = [ChartEncoding(channel="x", field="x"), ChartEncoding(channel="y", field="y")]
        dataset = DatasetSpec(dataset_id="data", columns=columns, rows=rows)
        chart = ChartSpec(chart_id="chart", family=family, dataset_id="data", encodings=encodings, x_label=x_label, y_label=y_label, interactions=["zoom", "hover", "select-datum", "linked-selection"])
        renderer = chart_renderer_for(chart)
        if grounded_rows:
            answer_markdown = f"I interpreted this as a **{family.replace('-', ' ')}** rendered through the verified {renderer.title()} adapter, using the values transcribed from your selected context."
        elif is_parabola:
            answer_markdown = "I rendered the standard parabola **y = x²** as an interactive 2D plot."
        else:
            answer_markdown = f"I interpreted this as a **{family.replace('-', ' ')}** and rendered it through the verified {renderer.title()} adapter." + (" The displayed values are illustrative because no explicit dataset was supplied." if illustrative else "")
        return CompositionSpec(
            project_id=project_id, prompt=prompt, title="Parabola" if is_parabola and not grounded_rows else f"{family.replace('-', ' ').title()}",
            answer_markdown=answer_markdown,
            parameters=[], datasets=[dataset], panels=[ChartPanelSpec(panel_id="chart", chart=chart, order=0)],
            assumptions=["The chart dataset is illustrative." ] if illustrative else [],
            provenance=ProvenanceSpec(
                interpretation=f"Semantic chart family: {family}" + (" (grounded in selected context)" if grounded_rows else ""),
                operator_sources=[renderer],
            ),
            assertions=[CompositionAssertion(assertion_id="dataset-shape", kind="dataset_shape", target_id="data")], created_at=time.time(),
        )