app/forge_bridge.py

"""Subprocess bridge to the vendored upstream ``forge/`` pipeline scripts.

This module is intentionally *standard-library only* so it can also run
inside the minimal Docker build stage that regenerates the test fixture
factory (``scripts/build_fixture_factory.py``).

Safety contract:
  * every invocation is a list-argv ``subprocess.run`` (never shell=True);
  * child processes get a scrubbed environment (no ``LLM_*``/``ANTHROPIC_*``
    variables can leak into them);
  * every call has a hard timeout and bounded captured output;
  * all file paths passed to the scripts live inside a per-job directory
    owned by the caller -- the scripts resolve and write whatever path they
    are given, so containment is enforced here by construction.
"""

from __future__ import annotations

import copy
import json
import math
import os
import re
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]
FORGE = REPO_ROOT / "forge"

PROBE = FORGE / "stage1_intake" / "probe_image.py"
VALIDATE = FORGE / "stage2_spec" / "validate_sculpt_spec.py"
ORCHESTRATE = FORGE / "stage3_build" / "orchestrate_passes.py"
GENERATE = FORGE / "stage3_build" / "generate_threejs_factory.py"

_MAX_CAPTURE = 1024 * 1024  # 1 MiB per stream


class ForgeError(Exception):
    """A forge script exited non-zero (or timed out). Carries diagnostics."""

    def __init__(self, script: str, message: str, *, returncode: int | None = None,
                 stdout: str = "", stderr: str = "") -> None:
        super().__init__(message)
        self.script = script
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr


@dataclass(frozen=True)
class ForgeResult:
    returncode: int
    stdout: str
    stderr: str


def _scrubbed_env() -> dict[str, str]:
    env = {
        "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"),
        "HOME": os.environ.get("HOME", "/tmp"),
        "PYTHONDONTWRITEBYTECODE": "1",
        "PYTHONUNBUFFERED": "1",
        "LANG": "C.UTF-8",
    }
    tmp = os.environ.get("TMPDIR")
    if tmp:
        env["TMPDIR"] = tmp
    return env


def run_forge(script: Path, *args: str, timeout: int = 120, cwd: Path | None = None) -> ForgeResult:
    """Run a forge script; return captured output. Never raises on rc != 0."""
    argv = [sys.executable, str(script), *args]
    try:
        proc = subprocess.run(
            argv,
            cwd=str(cwd) if cwd else None,
            env=_scrubbed_env(),
            capture_output=True,
            text=True,
            timeout=timeout,
            errors="replace",
        )
    except subprocess.TimeoutExpired as exc:
        raise ForgeError(
            script.name, f"{script.name} timed out after {timeout}s",
            returncode=None, stdout=exc.stdout or "", stderr=exc.stderr or "",
        ) from exc
    return ForgeResult(
        returncode=proc.returncode,
        stdout=proc.stdout[:_MAX_CAPTURE],
        stderr=proc.stderr[:_MAX_CAPTURE],
    )


def probe_image(image_path: Path, *, timeout: int = 60) -> dict:
    """Run stage1_intake/probe_image.py and parse its JSON stdout."""
    result = run_forge(PROBE, str(image_path), timeout=timeout)
    if result.returncode != 0:
        raise ForgeError(
            PROBE.name, f"probe_image failed (rc={result.returncode})",
            returncode=result.returncode, stdout=result.stdout, stderr=result.stderr,
        )
    try:
        payload = json.loads(result.stdout)
    except json.JSONDecodeError as exc:
        raise ForgeError(PROBE.name, "probe_image did not return JSON",
                         returncode=result.returncode, stdout=result.stdout,
                         stderr=result.stderr) from exc
    if not isinstance(payload, dict):
        raise ForgeError(PROBE.name, "probe_image returned non-object JSON",
                         returncode=result.returncode, stdout=result.stdout)
    return payload


def validate_spec(spec_path: Path, *, strict: bool = True, timeout: int = 60) -> dict:
    """Run the spec validator. Returns ``{ok, errors, warnings, ...}``.

    The validator exits 0 on PASS and 1 on FAIL; both are normal outcomes.
    rc > 1 (argparse/IO errors) raises ForgeError.
    """
    args = [str(spec_path), "--json"]
    if strict:
        args.append("--strict-quality")
    result = run_forge(VALIDATE, *args, timeout=timeout)
    if result.returncode not in (0, 1):
        raise ForgeError(
            VALIDATE.name, f"validate_sculpt_spec crashed (rc={result.returncode})",
            returncode=result.returncode, stdout=result.stdout, stderr=result.stderr,
        )
    try:
        payload = json.loads(result.stdout)
    except json.JSONDecodeError as exc:
        raise ForgeError(VALIDATE.name, "validator did not return JSON",
                         returncode=result.returncode, stdout=result.stdout,
                         stderr=result.stderr) from exc
    payload.setdefault("ok", result.returncode == 0)
    return payload


def orchestrate_sync(spec_path: Path, *, timeout: int = 60) -> None:
    """Refresh sculptPipeline from reviewHistory (sync --in-place)."""
    result = run_forge(ORCHESTRATE, "sync", str(spec_path), "--in-place", timeout=timeout)
    if result.returncode != 0:
        raise ForgeError(
            ORCHESTRATE.name, f"orchestrate sync failed (rc={result.returncode})",
            returncode=result.returncode, stdout=result.stdout, stderr=result.stderr,
        )


def generate_factory(spec_path: Path, out_path: Path, *, pass_id: str | None = None,
                     timeout: int = 120) -> None:
    """Emit the TypeScript factory. Raises ForgeError on gate/IO failure."""
    args = [str(spec_path), "--out", str(out_path), "--force"]
    if pass_id:
        args.extend(["--pass-id", pass_id])
    result = run_forge(GENERATE, *args, timeout=timeout)
    if result.returncode != 0:
        raise ForgeError(
            GENERATE.name,
            f"factory generation failed for pass {pass_id or '(current)'} "
            f"(rc={result.returncode}): {result.stderr.strip()[:400]}",
            returncode=result.returncode, stdout=result.stdout, stderr=result.stderr,
        )
    if not out_path.exists() or out_path.stat().st_size == 0:
        raise ForgeError(GENERATE.name, "generator exited 0 but produced no output",
                         returncode=result.returncode, stdout=result.stdout,
                         stderr=result.stderr)


HOSTED_PREVIEW_PASS = "hosted-preview"
HOSTED_PRIMITIVES = {
    "box", "sphere", "ellipsoid", "cylinder", "cone", "capsule", "torus",
    "tube", "lathe", "extrude", "ground-blade", "curve-sweep",
    "plane-card", "instanced-cluster",
}


def _is_finite_number(value: object) -> bool:
    """Return true only for real, finite JSON-style numbers.

    ``bool`` is an ``int`` subclass in Python, but accepting ``True`` for a
    radius or coordinate produces subtly malformed generated JavaScript.
    Very large Python integers may also overflow during float conversion, so
    keep the conversion guarded.
    """
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return False
    try:
        return math.isfinite(float(value))
    except (OverflowError, ValueError):
        return False


def _is_positive_number(value: object) -> bool:
    return _is_finite_number(value) and float(value) > 0


def _is_integer_at_least(value: object, minimum: int) -> bool:
    return (
        _is_finite_number(value)
        and float(value).is_integer()
        and float(value) >= minimum
    )


def _is_coordinate(value: object, size: int) -> bool:
    return (
        isinstance(value, (list, tuple))
        and len(value) == size
        and all(_is_finite_number(item) for item in value)
    )


def _valid_point_array(value: object, size: int, minimum: int) -> bool:
    if (
        not isinstance(value, list)
        or len(value) < minimum
        or not all(_is_coordinate(point, size) for point in value)
    ):
        return False
    # Degenerate paths/polygons with repeated copies of one coordinate satisfy
    # a simple length check but still fail inside Three.js geometry builders.
    points = {tuple(float(item) for item in point) for point in value}
    return len(points) >= minimum


def _advanced_geometry_errors(
    label: str, primitive: str, descriptor: dict
) -> list[str]:
    """Validate descriptor payloads consumed directly by generated TypeScript."""
    errors: list[str] = []
    prefix = f"hosted preview: component {label!r} geometryDescriptor"

    if primitive == "extrude":
        profile = descriptor.get("profile2D")
        field = f"{prefix}.profile2D"
        if not isinstance(profile, dict):
            return [f"{field} must be an object"]
        if not _valid_point_array(profile.get("points"), 2, 3):
            errors.append(
                f"{field}.points must contain at least 3 distinct finite [x, y] coordinates"
            )
        if not _is_positive_number(profile.get("depth")):
            errors.append(f"{field}.depth must be a finite positive number")
        holes = profile.get("holes")
        if holes is not None and (
            not isinstance(holes, list)
            or not all(_valid_point_array(loop, 2, 3) for loop in holes)
        ):
            errors.append(
                f"{field}.holes must contain polygons of at least 3 distinct finite "
                "[x, y] coordinates"
            )
        oval_holes = profile.get("ovalHoles")
        if oval_holes is not None:
            valid_ovals = isinstance(oval_holes, list) and all(
                isinstance(oval, dict)
                and _is_finite_number(oval.get("cx"))
                and _is_finite_number(oval.get("cy"))
                and _is_positive_number(oval.get("rx"))
                and _is_positive_number(oval.get("ry"))
                for oval in oval_holes
            )
            if not valid_ovals:
                errors.append(
                    f"{field}.ovalHoles must contain finite cx/cy and positive finite rx/ry"
                )

    elif primitive == "lathe":
        profile = descriptor.get("latheProfile")
        field = f"{prefix}.latheProfile"
        if not isinstance(profile, dict):
            return [f"{field} must be an object"]
        points = profile.get("points")
        if not _valid_point_array(points, 2, 2):
            errors.append(
                f"{field}.points must contain at least 2 distinct finite [radius, y] coordinates"
            )
        elif any(float(point[0]) < 0 for point in points):
            errors.append(f"{field}.points radii must be finite non-negative numbers")
        if not _is_integer_at_least(profile.get("segments"), 3):
            errors.append(f"{field}.segments must be a finite integer of at least 3")

    elif primitive == "tube":
        path = descriptor.get("tubePath")
        field = f"{prefix}.tubePath"
        if not isinstance(path, dict):
            return [f"{field} must be an object"]
        if not _valid_point_array(path.get("points"), 3, 2):
            errors.append(
                f"{field}.points must contain at least 2 distinct finite [x, y, z] coordinates"
            )
        if not _is_positive_number(path.get("radius")):
            errors.append(f"{field}.radius must be a finite positive number")
        if "radialSegments" in path and not _is_integer_at_least(
            path["radialSegments"], 3
        ):
            errors.append(
                f"{field}.radialSegments must be a finite integer of at least 3"
            )
        if not isinstance(path.get("closed"), bool):
            errors.append(f"{field}.closed must be boolean")

    elif primitive == "ground-blade":
        blade = descriptor.get("bladeSpec")
        field = f"{prefix}.bladeSpec"
        if not isinstance(blade, dict):
            return [f"{field} must be an object"]
        stations = blade.get("stations")
        if not _valid_point_array(stations, 3, 2):
            errors.append(
                f"{field}.stations must contain at least 2 distinct finite "
                "[x, spineY, edgeY] coordinates"
            )
        elif any(float(station[1]) <= float(station[2]) for station in stations):
            errors.append(f"{field}.stations must keep spineY greater than edgeY")
        elif any(
            float(stations[index][0]) <= float(stations[index - 1][0])
            for index in range(1, len(stations))
        ):
            errors.append(f"{field}.stations x coordinates must be strictly increasing")
        if not _is_positive_number(blade.get("thickness")):
            errors.append(f"{field}.thickness must be a finite positive number")
        fraction_defaults = {
            "grindFrac": 0.55,
            "swedgeFromTipFrac": 0.34,
            "spineFlat": 0.30,
        }
        for fraction, default in fraction_defaults.items():
            value = blade.get(fraction, default)
            if not (
                _is_finite_number(value)
                and float(value) >= 0
                and float(value) <= 1
            ):
                errors.append(
                    f"{field}.{fraction} must be a finite number from 0 to 1"
                )
        grind = blade.get("grindFrac", fraction_defaults["grindFrac"])
        spine_flat = blade.get("spineFlat", fraction_defaults["spineFlat"])
        if (
            _is_finite_number(grind)
            and 0 <= float(grind) <= 1
            and _is_finite_number(spine_flat)
            and 0 <= float(spine_flat) <= 1
            and float(grind) + float(spine_flat) >= 1
        ):
            errors.append(f"{field}.grindFrac + spineFlat must be less than 1")

    elif primitive == "curve-sweep":
        sweep = descriptor.get("curveSweep")
        field = f"{prefix}.curveSweep"
        if not isinstance(sweep, dict):
            return [f"{field} must be an object"]
        if not _valid_point_array(sweep.get("spine"), 3, 2):
            errors.append(
                f"{field}.spine must contain at least 2 distinct finite [x, y, z] coordinates"
            )
        cross_section = sweep.get("crossSection")
        if not isinstance(cross_section, dict):
            errors.append(f"{field}.crossSection must be an object")
        elif not _valid_point_array(cross_section.get("points"), 2, 3):
            errors.append(
                f"{field}.crossSection.points must contain at least 3 distinct finite "
                "[x, y] coordinates"
            )
        if not isinstance(sweep.get("closed"), bool):
            errors.append(f"{field}.closed must be boolean")

    return errors


def _resolved_dimensions(component: dict) -> tuple[object, object, object]:
    dimensions = component.get("dimensions")
    if not isinstance(dimensions, dict):
        return None, None, None
    radius = dimensions.get("radius")
    diameter = float(radius) * 2 if _is_positive_number(radius) else None
    width = dimensions.get("width", diameter)
    height = dimensions.get("height", dimensions.get("length"))
    depth = dimensions.get("depth", diameter)
    if component.get("primitive") == "plane-card" and depth is None:
        depth = 1.0
    return width, height, depth


def hosted_spec_errors(spec: dict) -> list[str]:
    """Return deterministic limitations of the hosted preview compiler.

    The upstream schema accepts procedural primitive families that the
    vendored generator still represents with placeholder boxes. The hosted
    service rejects those before generation rather than returning a model
    that silently disagrees with its spec. Parent cycles are also rejected
    explicitly so generator recursion cannot hang.
    """
    components = [
        item for item in spec.get("componentTree", []) if isinstance(item, dict)
    ]
    errors: list[str] = []
    ids = {
        str(item.get("id")) for item in components
        if isinstance(item.get("id"), str) and item.get("id")
    }
    parents: dict[str, str | None] = {}
    for index, component in enumerate(components):
        component_id = component.get("id")
        label = str(component_id or f"componentTree[{index}]")
        primitive = str(component.get("primitive") or "")
        if primitive not in HOSTED_PRIMITIVES:
            errors.append(
                f"hosted preview: component {label!r} uses unsupported primitive "
                f"{primitive!r}; choose one of {', '.join(sorted(HOSTED_PRIMITIVES))}"
            )
        dimensions = _resolved_dimensions(component)
        if not all(_is_positive_number(value) for value in dimensions):
            errors.append(
                f"hosted preview: component {label!r} needs positive width, height, "
                "and depth dimensions (radius/length aliases are accepted)"
            )
        if primitive in {
            "extrude", "lathe", "tube", "ground-blade", "curve-sweep",
            "instanced-cluster",
        }:
            descriptor = component.get("geometryDescriptor")
            if not isinstance(descriptor, dict):
                errors.append(
                    f"hosted preview: component {label!r} geometryDescriptor must be an object"
                )
            elif primitive == "instanced-cluster":
                base = descriptor.get("baseGeometry")
                if not isinstance(base, str) or not base:
                    errors.append(
                        f"hosted preview: component {label!r} "
                        "geometryDescriptor.baseGeometry must name a supported primitive"
                    )
                elif base == "instanced-cluster":
                    errors.append(
                        f"hosted preview: component {label!r} "
                        "geometryDescriptor.baseGeometry cannot reference instanced-cluster itself"
                    )
                elif base not in HOSTED_PRIMITIVES:
                    errors.append(
                        f"hosted preview: component {label!r} "
                        f"geometryDescriptor.baseGeometry uses unsupported primitive {base!r}"
                    )
                else:
                    errors.extend(
                        _advanced_geometry_errors(label, base, descriptor)
                    )
            else:
                errors.extend(
                    _advanced_geometry_errors(label, primitive, descriptor)
                )
        if isinstance(component_id, str) and component_id:
            parent = component.get("parent")
            parents[component_id] = str(parent) if parent is not None else None

    for component_id, parent in parents.items():
        if parent is not None and parent not in ids:
            errors.append(
                f"hosted preview: component {component_id!r} references missing parent {parent!r}"
            )
            continue
        seen: set[str] = set()
        cursor: str | None = component_id
        while cursor is not None and cursor in parents:
            if cursor in seen:
                errors.append(
                    f"hosted preview: component parent cycle contains {cursor!r}"
                )
                break
            seen.add(cursor)
            cursor = parents[cursor]
    return list(dict.fromkeys(errors))


def prepare_hosted_preview(spec: dict) -> dict:
    """Create an app-only compile manifest without claiming visual approval.

    ``spec`` remains the strict-gated upstream artifact with its locked pass
    order and empty review history. The returned deep copy contains one
    explicitly unreviewed preview pass referencing every declared component.
    It is compiled solely to let users inspect the procedural draft; it is
    not an upstream ``continue`` decision and carries no scores or evidence.
    """
    preview = copy.deepcopy(spec)
    component_refs = [
        item["id"] for item in preview.get("componentTree", [])
        if isinstance(item, dict) and isinstance(item.get("id"), str) and item["id"]
    ]
    source_order = pass_order(spec)
    preview["buildPasses"] = [{
        "id": HOSTED_PREVIEW_PASS,
        "label": "Hosted unreviewed preview",
        "goal": "Compile all strict-validated components for inspection only",
        "componentRefs": component_refs,
        "acceptance": ["No visual acceptance is claimed by this preview"],
        "acceptanceCriteria": ["No visual acceptance is claimed by this preview"],
    }]
    preview["sculptPipeline"] = {
        "passGateMode": "hosted-unreviewed-preview",
        "passOrder": [HOSTED_PREVIEW_PASS],
        "currentPass": HOSTED_PREVIEW_PASS,
        "completedPasses": [],
        "sourcePassOrder": source_order,
    }
    preview["reviewHistory"] = []
    preview["hostedPreview"] = {
        "reviewStatus": "unreviewed",
        "sourcePassOrder": source_order,
        "notice": (
            "Compile-only preview. No screenshot comparison, AI-vision score, "
            "or upstream build-pass approval has occurred."
        ),
    }
    return preview


def pass_order(spec: dict) -> list[str]:
    ids = [
        item["id"]
        for item in spec.get("buildPasses", [])
        if isinstance(item, dict) and isinstance(item.get("id"), str) and item["id"].strip()
    ]
    return ids or ["blockout"]


_FACTORY_EXPORT_RE = re.compile(r"export function (create\w+Model)\b")


def factory_export_name(ts_source: str) -> str | None:
    """Extract the ``create<Name>Model`` export from generated TypeScript."""
    match = _FACTORY_EXPORT_RE.search(ts_source)
    return match.group(1) if match else None