app/pipeline.py

"""Job orchestration: image in -> ObjectSculptSpec -> Three.js factory -> bundle.

This preserves the upstream repository's intended conversion flow
(image -> probe -> LLM-authored ObjectSculptSpec -> strict-quality gate ->
an inspectable Three.js preview) and adapts the interactive agent loop to a
hosted request/response service:

  * the LLM (vision) authors the spec, exactly as the skill intends;
  * validate_sculpt_spec.py --strict-quality gates it; validator and hosted
    compiler errors are fed back for up to ``spec_repair_rounds`` repairs;
  * the original spec remains locked and unreviewed. A separate compile-only
    manifest gathers its validated components into an explicitly unreviewed
    hosted preview pass; it contains no invented scores or screenshot paths;
  * the emitted TypeScript factory is bundled (esbuild, three included) into
    a single ESM artifact plus a self-contained standalone HTML export.

Every failure path returns an honest error; nothing is ever fabricated.
"""

from __future__ import annotations

import asyncio
import base64
import json
import os
import re
import shutil
import sys
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from . import forge_bridge
from .config import Settings
from .gallery import GalleryError, GalleryStore
from .image_guard import ImageRejected, validate_and_normalize
from .llm import LLMClient, LLMError, extract_json_object
from .prompt import build_repair_prompt, build_system_prompt, build_user_prompt

REPO_ROOT = Path(__file__).resolve().parents[1]
STATIC_DIR = Path(__file__).resolve().parent / "static"

TARGET_NAME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9 ]{0,38}$")
MAX_SPEC_BYTES = 512 * 1024
LLM_PROGRESS_INTERVAL_S = 25.0
STAGE_PROGRESS_INTERVAL_S = 25.0

ARTIFACT_NAMES = {
    "reference.png": "image/png",
    "probe.json": "application/json",
    "spec.json": "application/json",
    "compile-spec.json": "application/json",
    "validation.json": "application/json",
    "factory.ts": "text/plain; charset=utf-8",
    "model.bundle.js": "text/javascript; charset=utf-8",
    "standalone.html": "text/html; charset=utf-8",
    "events.jsonl": "application/x-ndjson; charset=utf-8",
}


class PipelineError(Exception):
    """Honest, user-presentable pipeline failure."""

    def __init__(self, message: str, *, code: str = "pipeline_error",
                 stage: str = "", detail: Any = None) -> None:
        super().__init__(message)
        self.code = code
        self.stage = stage
        self.detail = detail


@dataclass
class Job:
    id: str
    dir: Path
    created: float = field(default_factory=time.time)
    finished: float | None = None
    status: str = "running"  # running | done | error
    stage: str = "queued"
    seq: int = 0
    events: list[dict] = field(default_factory=list)
    result: dict | None = None
    error: dict | None = None
    waiter: asyncio.Event = field(default_factory=asyncio.Event)

    def emit(self, stage: str, status: str, message: str, **data: Any) -> dict:
        self.seq += 1
        self.stage = stage
        event = {
            "seq": self.seq,
            "ts": round(time.time(), 3),
            "stage": stage,
            "status": status,  # started | progress | done | error
            "message": message,
        }
        if data:
            event["data"] = data
        self.events.append(event)
        try:
            with (self.dir / "events.jsonl").open("a", encoding="utf-8") as fh:
                fh.write(json.dumps(event) + "\n")
        except OSError:
            pass
        self.waiter.set()
        return event


class JobRegistry:
    def __init__(self, runs_dir: Path) -> None:
        self.runs_dir = runs_dir
        self.jobs: dict[str, Job] = {}

    def create(self) -> Job:
        job_id = uuid.uuid4().hex
        job_dir = self.runs_dir / job_id
        job_dir.mkdir(parents=True, exist_ok=False)
        job = Job(id=job_id, dir=job_dir)
        self.jobs[job_id] = job
        return job

    def get(self, job_id: str) -> Job | None:
        if not re.fullmatch(r"[0-9a-f]{32}", job_id or ""):
            return None
        return self.jobs.get(job_id)

    def evict(self, job_id: str) -> None:
        job = self.jobs.pop(job_id, None)
        if job is not None:
            shutil.rmtree(job.dir, ignore_errors=True)

    def reap(self, ttl_s: int) -> list[str]:
        now = time.time()
        doomed = [
            j.id for j in self.jobs.values()
            if j.finished is not None and now - j.finished > ttl_s
        ]
        for job_id in doomed:
            self.evict(job_id)
        # Also sweep orphaned directories (e.g. after a crash).
        if self.runs_dir.exists():
            known = {j.dir for j in self.jobs.values()}
            for child in self.runs_dir.iterdir():
                if (
                    child.is_dir()
                    and child not in known
                    and re.fullmatch(r"[0-9a-f]{32}", child.name)
                ):
                    try:
                        if now - child.stat().st_mtime > ttl_s:
                            shutil.rmtree(child, ignore_errors=True)
                    except OSError:
                        pass
        return doomed


def sanitize_spec(spec: dict) -> dict:
    """Clamp LLM-authored values that the generator/validator treat strictly."""
    name = spec.get("targetName")
    if not isinstance(name, str) or not TARGET_NAME_RE.fullmatch(name.strip()):
        spec["targetName"] = "Object"
    else:
        spec["targetName"] = name.strip()
    spec["schemaVersion"] = "2.1"
    suitability = spec.get("suitability")
    if suitability not in {"pass", "conditional", "reject"}:
        spec["suitability"] = "conditional"
    spec["reviewHistory"] = []  # LLM must not pre-approve its own passes
    return spec


async def _complete_vision_with_feedback(
    job: Job,
    *,
    llm: LLMClient,
    system: str,
    messages: list[dict],
    attempt: int,
    max_attempts: int,
):
    """Await one provider call while reporting truthful, bounded-cadence waits.

    Provider APIs do not expose a meaningful percentage, so the heartbeat only
    reports elapsed wait time and the current validation attempt. The child
    task is always cancelled if the job timeout or caller cancels this await.
    """
    waiting_since = time.monotonic()
    task = asyncio.create_task(
        llm.complete_vision(system=system, messages=messages)
    )
    try:
        while True:
            done, _ = await asyncio.wait(
                {task}, timeout=LLM_PROGRESS_INTERVAL_S
            )
            if task in done:
                return task.result()
            elapsed = max(1, round(time.monotonic() - waiting_since))
            job.emit(
                "spec-authoring",
                "progress",
                f"Still waiting for the vision model ({elapsed}s elapsed, "
                f"attempt {attempt} of {max_attempts}).",
                attempt=attempt,
                maxAttempts=max_attempts,
                elapsedSeconds=elapsed,
            )
    finally:
        if not task.done():
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass


async def _await_with_feedback(
    job: Job,
    *,
    stage: str,
    awaitable,
    message: str,
    **data: Any,
):
    """Await one opaque stage while emitting elapsed-only progress.

    Generator, bundler, and Bucket APIs expose no honest percentage. A bounded
    cadence with stage and total elapsed time gives useful feedback without
    inventing completion estimates.
    """
    stage_started = time.monotonic()
    task = asyncio.ensure_future(awaitable)
    try:
        while True:
            done, _ = await asyncio.wait(
                {task}, timeout=STAGE_PROGRESS_INTERVAL_S
            )
            if task in done:
                return task.result()
            stage_elapsed = max(1, round(time.monotonic() - stage_started))
            total_elapsed = max(1, round(time.time() - job.created))
            job.emit(
                stage,
                "progress",
                message.format(elapsed=stage_elapsed, total=total_elapsed),
                stageElapsedSeconds=stage_elapsed,
                elapsedSeconds=total_elapsed,
                **data,
            )
    finally:
        if not task.done():
            task.cancel()
            try:
                await task
            except (asyncio.CancelledError, Exception):
                pass


def _gallery_worker_command(request_path: Path, response_path: Path) -> list[str]:
    return [
        sys.executable,
        "-m",
        "app.gallery_worker",
        str(request_path),
        str(response_path),
    ]


async def _publish_gallery(
    store: GalleryStore,
    *,
    job: Job,
    result: dict[str, Any],
    created_at: float,
) -> dict[str, Any]:
    """Publish with a process boundary so cancellation can stop Bucket I/O."""
    # Tests may inject a behavioural subclass. Production uses the exact
    # GalleryStore and therefore always takes the killable process path.
    if type(store) is not GalleryStore:
        return await asyncio.to_thread(
            store.publish,
            item_id=job.id,
            job_dir=job.dir,
            result=result,
            created_at=created_at,
            elapsed_started_at=job.created,
        )

    token = uuid.uuid4().hex
    request_path = job.dir / f".gallery-publish-{token}.request.json"
    response_path = job.dir / f".gallery-publish-{token}.response.json"
    request_path.write_text(
        json.dumps(
            {
                "galleryDir": str(store.root),
                "itemId": job.id,
                "jobDir": str(job.dir),
                "result": result,
                "createdAt": created_at,
                "elapsedStartedAt": job.created,
                "stagingToken": token,
            },
            separators=(",", ":"),
        ),
        encoding="utf-8",
    )

    environment = {
        "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"),
        "PYTHONPATH": str(REPO_ROOT),
        "PYTHONUNBUFFERED": "1",
    }
    proc = await asyncio.create_subprocess_exec(
        *_gallery_worker_command(request_path, response_path),
        cwd=str(REPO_ROOT),
        env=environment,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    try:
        try:
            await proc.communicate()
        except BaseException:
            if proc.returncode is None:
                proc.terminate()
                try:
                    await asyncio.wait_for(proc.wait(), timeout=2)
                except asyncio.TimeoutError:
                    proc.kill()
                    try:
                        await asyncio.wait_for(proc.wait(), timeout=3)
                    except (asyncio.TimeoutError, ProcessLookupError):
                        pass
                except ProcessLookupError:
                    pass
            raise

        if not response_path.is_file():
            raise GalleryError(
                "The isolated community gallery publisher exited without a result."
            )
        try:
            response = json.loads(response_path.read_text(encoding="utf-8"))
        except (OSError, UnicodeError, json.JSONDecodeError) as exc:
            raise GalleryError(
                "The isolated community gallery publisher returned an invalid result."
            ) from exc
        if proc.returncode != 0 or response.get("ok") is not True:
            raise GalleryError(
                str(response.get("error") or "Community gallery publication failed.")
            )
        item = response.get("item")
        if not isinstance(item, dict) or item.get("id") != job.id:
            raise GalleryError(
                "The isolated community gallery publisher returned the wrong item."
            )
        return item
    finally:
        request_path.unlink(missing_ok=True)
        response_path.unlink(missing_ok=True)


async def run_job(
    job: Job,
    *,
    raw_upload: bytes,
    object_hint: str | None,
    settings: Settings,
    llm: LLMClient,
    share: bool = True,
    gallery_store: GalleryStore | None = None,
) -> None:
    """Execute the full conversion for one job. Never raises: failures are
    recorded on the job and emitted as terminal error events."""
    started = job.created
    remaining = max(0.0, settings.job_timeout_s - (time.time() - started))
    try:
        async with asyncio.timeout(remaining) as deadline:
            await _run(job, raw_upload=raw_upload, object_hint=object_hint,
                       settings=settings, llm=llm, started=started,
                       share=share, gallery_store=gallery_store,
                       deadline=deadline)
    except TimeoutError:
        fail_job_timeout(job, settings.job_timeout_s)
    except ImageRejected as exc:
        _fail(job, stage="intake", code=exc.code, message=exc.reason)
    except LLMError as exc:
        _fail(job, stage="spec-authoring", code=exc.code, message=str(exc))
    except PipelineError as exc:
        _fail(job, stage=exc.stage or job.stage, code=exc.code, message=str(exc),
              detail=exc.detail)
    except Exception as exc:  # pragma: no cover - defensive catch-all
        _fail(job, stage=job.stage, code="internal_error",
              message=f"Unexpected internal error: {type(exc).__name__}. "
                      "Nothing was generated.")


async def _run(job: Job, *, raw_upload: bytes, object_hint: str | None,
               settings: Settings, llm: LLMClient, started: float,
               share: bool, gallery_store: GalleryStore | None,
               deadline: asyncio.Timeout) -> None:
    # -- stage 1: intake -----------------------------------------------------
    job.emit("intake", "started", "Validating and normalising the uploaded image.")
    normalized = await asyncio.to_thread(
        validate_and_normalize, raw_upload,
        max_bytes=settings.max_upload_bytes,
        max_pixels=settings.max_image_pixels,
        normalize_max_side=settings.normalize_max_side,
    )
    reference = job.dir / "reference.png"
    reference.write_bytes(normalized.png_bytes)
    probe = await asyncio.to_thread(forge_bridge.probe_image, reference)
    (job.dir / "probe.json").write_text(json.dumps(probe, indent=2), encoding="utf-8")
    job.emit(
        "intake", "done",
        f"Image accepted: {normalized.original_format} "
        f"{normalized.original_width}x{normalized.original_height}px"
        + (" (downscaled for processing)" if normalized.downscaled else ""),
        probe={"width": probe.get("width"), "height": probe.get("height"),
               "warnings": probe.get("warnings", [])},
    )

    # -- stage 2: LLM authors the ObjectSculptSpec ---------------------------
    system = build_system_prompt()
    from .llm import assistant_turn, user_turn
    messages: list[dict] = []
    spec: dict | None = None
    validation: dict = {}
    repair_rounds = max(0, settings.spec_repair_rounds)
    max_attempts = 1 + repair_rounds
    for attempt in range(1, max_attempts + 1):
        if attempt == 1:
            job.emit(
                "spec-authoring", "started",
                f"The vision model ({settings.llm_model}) is studying the image and "
                "authoring the ObjectSculptSpec (components, materials, proportions).",
                attempt=attempt, maxAttempts=max_attempts,
            )
            messages.append(user_turn(
                build_user_prompt(probe, object_hint=object_hint),
                image_png=normalized.png_bytes,
            ))
        else:
            job.emit(
                "spec-authoring", "progress",
                f"Validator rejected the spec; the model is repairing it "
                f"(attempt {attempt} of {max_attempts}).",
                attempt=attempt, maxAttempts=max_attempts,
                validatorErrors=(validation.get("errors") or [])[:8],
            )
            messages.append(user_turn(build_repair_prompt(
                validation.get("errors") or [], validation.get("warnings") or [])))

        reply = await _complete_vision_with_feedback(
            job,
            llm=llm,
            system=system,
            messages=messages,
            attempt=attempt,
            max_attempts=max_attempts,
        )
        messages.append(assistant_turn(reply.text))
        try:
            candidate = extract_json_object(reply.text)
        except LLMError as exc:
            if exc.code != "llm_bad_json":
                raise
            validation = {
                "ok": False,
                "errors": [
                    "The model response was not one complete parseable JSON object."
                ],
                "warnings": [],
            }
            (job.dir / "validation.json").write_text(
                json.dumps(validation, indent=2), encoding="utf-8")
            continue
        if len(json.dumps(candidate)) > MAX_SPEC_BYTES:
            raise PipelineError(
                "The model produced an unreasonably large spec (>512 KB). "
                "Try a simpler subject or crop the image.",
                code="spec_too_large", stage="spec-authoring")
        candidate = sanitize_spec(candidate)

        # Suitability "reject" is an honest, valid pipeline outcome.
        if candidate.get("suitability") == "reject":
            (job.dir / "spec.json").write_text(json.dumps(candidate, indent=2),
                                               encoding="utf-8")
            raise PipelineError(
                "The model judged this image unsuitable for a faithful procedural "
                "reconstruction (suitability: reject). Try a single object with a "
                "clear silhouette on a plain background. No model was generated.",
                code="unsuitable_image", stage="spec-authoring",
                detail={"specUrl": f"/api/jobs/{job.id}/artifacts/spec.json"})

        spec_path = job.dir / "spec.json"
        spec_path.write_text(json.dumps(candidate, indent=2), encoding="utf-8")
        validation = await asyncio.to_thread(
            forge_bridge.validate_spec, spec_path, strict=True)
        hosted_errors = forge_bridge.hosted_spec_errors(candidate)
        if hosted_errors:
            validation["ok"] = False
            validation["errors"] = list(validation.get("errors") or []) + hosted_errors
        (job.dir / "validation.json").write_text(json.dumps(validation, indent=2),
                                                 encoding="utf-8")
        if validation.get("ok"):
            spec = candidate
            job.emit(
                "spec-authoring", "done",
                f"Spec accepted by the strict-quality gate on round {attempt}: "
                f"{candidate.get('targetName', 'Object')} — "
                f"{len(candidate.get('componentTree', []))} components, "
                f"{len(candidate.get('materials', []))} materials.",
                attempt=attempt,
                warnings=validation.get("warnings") or [],
            )
            break
    if spec is None:
        raise PipelineError(
            f"The spec failed the deterministic strict-quality gate after "
            f"{max_attempts} attempts ({repair_rounds} repair rounds). This is the "
            "gate working as designed, not a crash. "
            "No model was generated.",
            code="spec_validation_failed", stage="spec-authoring",
            detail={"errors": (validation.get("errors") or [])[:20],
                    "validationUrl": f"/api/jobs/{job.id}/artifacts/validation.json"})

    # -- stage 3: honest compile-only hosted preview -------------------------
    order = forge_bridge.pass_order(spec)
    job.emit("generation", "started",
             "Strict gate passed. Preparing an unreviewed procedural preview; "
             "the upstream pass order remains locked pending real screenshots "
             f"and visual review ({' -> '.join(order)}).")

    ts_path = job.dir / "factory.ts"
    generated_pass = forge_bridge.HOSTED_PREVIEW_PASS
    preview_spec = forge_bridge.prepare_hosted_preview(spec)
    preview_path = job.dir / "compile-spec.json"
    preview_path.write_text(json.dumps(preview_spec, indent=2), encoding="utf-8")
    try:
        await _await_with_feedback(
            job,
            stage="generation",
            awaitable=asyncio.to_thread(
                forge_bridge.generate_factory,
                preview_path,
                ts_path,
                pass_id=generated_pass,
            ),
            message=(
                "The procedural generator is still compiling the validated "
                "spec ({elapsed}s in this stage, {total}s total)."
            ),
            generatedPass=generated_pass,
        )
    except forge_bridge.ForgeError as exc:
        raise PipelineError(
            "The strict-validated spec could not be compiled into the hosted "
            f"preview ({exc.stderr.strip()[:300] or 'generator refusal'}). "
            "No model was generated.",
            code="generation_failed", stage="generation",
        ) from exc

    ts_source = ts_path.read_text(encoding="utf-8")
    if "TODO:" in ts_source:
        raise PipelineError(
            "The generator attempted to emit placeholder geometry. The preview "
            "was refused instead of returning a fabricated shape.",
            code="generation_failed", stage="generation",
        )
    export_name = forge_bridge.factory_export_name(ts_source)
    if not export_name:
        raise PipelineError(
            "The generated factory has no create<Name>Model export. "
            "No model was generated.",
            code="generation_failed", stage="generation")
    job.emit(
        "generation",
        "done",
        f"Procedural factory emitted and verified with export {export_name}.",
        generatedPass=generated_pass,
        exportName=export_name,
    )

    # -- stage 4: bundle for the browser (esbuild) ---------------------------
    job.emit("bundling", "started",
             "Unreviewed preview factory emitted. Bundling with three.js "
             "for the in-browser viewer.")
    entry = job.dir / "entry.js"
    pascal = re.sub(r"^create|Model$", "", export_name)
    entry.write_text(
        f'export {{ {export_name} as makeModel, '
        f'create{pascal}LookDevLights as makeLights }} from "./factory.ts";\n'
        f'export {{ mountViewer }} from {json.dumps(str(STATIC_DIR / "viewer-core.js"))};\n',
        encoding="utf-8")
    bundle_path = job.dir / "model.bundle.js"
    await _await_with_feedback(
        job,
        stage="bundling",
        awaitable=_run_esbuild(job, entry, bundle_path, settings),
        message=(
            "The browser bundle is still being assembled "
            "({elapsed}s in this stage, {total}s total)."
        ),
    )

    # Self-contained standalone export (works offline, single file).
    bundle_text = bundle_path.read_text(encoding="utf-8")
    standalone = _build_standalone(bundle_text, spec.get("targetName", "Object"))
    (job.dir / "standalone.html").write_text(standalone, encoding="utf-8")

    job.emit("bundling", "done", "Browser bundle and standalone export ready.")

    # The bounded conversion is complete and every downloadable artifact
    # exists. Optional Bucket publication has a separate, shorter process
    # deadline below so a storage outage cannot consume the worker forever.
    deadline.reschedule(None)

    # -- stage 5: apply the explicit community sharing choice -----------------
    elapsed = round(time.time() - started, 1)
    components = spec.get("componentTree", [])
    result = {
        "jobId": job.id,
        "targetName": spec.get("targetName", "Object"),
        "generatedPass": generated_pass,
        "generationMode": "hosted-unreviewed-preview",
        "reviewStatus": "unreviewed",
        "passOrder": order,
        "completedPasses": [],
        "components": len(components),
        "materials": len(spec.get("materials", [])),
        "validationWarnings": validation.get("warnings") or [],
        "elapsedSeconds": elapsed,
        "shareRequested": share,
        "shared": False,
        "galleryItem": None,
        "artifacts": {name: f"/api/jobs/{job.id}/artifacts/{name}"
                      for name in ARTIFACT_NAMES if (job.dir / name).exists()},
        "honesty": [
            "Approximate procedural reconstruction from one image; hidden geometry "
            "is a model inference, not an observation or measurement.",
            "This is an unreviewed hosted preview. The original spec has no pass "
            "approvals, screenshot comparisons, or visual-fidelity scores.",
            "Use the upstream render/comparison/review loop before treating any "
            "build pass as visually accepted.",
            "The factory is the upstream generator's procedural scaffold: "
            "proportions, materials and structure come from the LLM-authored spec; "
            "fine surface artistry is out of scope for v1.",
        ],
    }
    if share:
        job.emit(
            "publishing",
            "started",
            "The finished result is being copied into the persistent community gallery.",
            shareRequested=True,
        )
        store = gallery_store or GalleryStore(Path(settings.gallery_dir))
        try:
            async with asyncio.timeout(settings.gallery_publish_timeout_s):
                gallery_item = await _await_with_feedback(
                    job,
                    stage="publishing",
                    awaitable=_publish_gallery(
                        store,
                        job=job,
                        result=result,
                        created_at=time.time(),
                    ),
                    message=(
                        "The persistent gallery copy is still being committed "
                        "({elapsed}s in this stage, {total}s total)."
                    ),
                    shareRequested=True,
                )
        except TimeoutError:
            publication_error = GalleryError(
                "Community gallery publication exceeded its storage deadline."
            )
        except GalleryError as exc:
            publication_error = exc
        else:
            publication_error = None

        if publication_error is not None:
            # Publication is an optional copy made after every model artifact
            # is complete. A storage outage must not discard an otherwise
            # usable result or hide its viewer/download controls.
            warning = (
                "The model is ready, but its requested community gallery copy "
                "could not be committed. The incomplete copy was discarded; "
                "download this temporary result before its job expires."
            )
            result["publicationWarning"] = warning
            job.emit(
                "publishing",
                "done",
                warning,
                shareRequested=True,
                shared=False,
                warning=True,
            )
        else:
            result["shared"] = True
            result["galleryItem"] = gallery_item
            job.emit(
                "publishing",
                "done",
                "Published to the community gallery.",
                shareRequested=True,
                shared=True,
                galleryItem=gallery_item,
            )
    else:
        job.emit(
            "publishing",
            "done",
            "Community sharing was turned off; this result remains in temporary job storage.",
            shareRequested=False,
            shared=False,
        )

    # -- done ------------------------------------------------------------------
    elapsed = round(time.time() - started, 1)
    result["elapsedSeconds"] = elapsed
    job.result = result
    job.status = "done"
    job.finished = time.time()
    job.emit("done", "done",
             f"Done in {elapsed}s — {result['targetName']} "
             f"({result['components']} components, pass '{generated_pass}').",
             result=result)


def _fail(job: Job, *, stage: str, code: str, message: str,
          detail: Any = None) -> None:
    if job.status != "running":
        return
    job.status = "error"
    job.finished = time.time()
    job.error = {"code": code, "message": message, "stage": stage,
                 **({"detail": detail} if detail else {})}
    job.emit(stage, "error", message, code=code, **({"detail": detail} if detail else {}))


def fail_job_timeout(job: Job, timeout_s: float) -> None:
    if timeout_s < 60:
        deadline = f"{max(1, round(timeout_s))}-second"
    else:
        deadline = f"{max(1, round(timeout_s / 60))}-minute"
    _fail(
        job,
        stage=job.stage,
        code="job_timeout",
        message=(
            f"The conversion exceeded its total {deadline} queue-and-build "
            "deadline and was stopped. No unverified model was returned."
        ),
    )


async def _run_esbuild(job: Job, entry: Path, out: Path, settings: Settings) -> None:
    esbuild = REPO_ROOT / settings.esbuild_entry
    if not esbuild.exists():
        raise PipelineError(
            "The esbuild bundler is missing from this deployment (build misconfiguration).",
            code="bundler_missing", stage="bundling")
    # Let esbuild resolve `three` from the image's node_modules.
    link = job.dir / "node_modules"
    if not link.exists():
        try:
            link.symlink_to(REPO_ROOT / "node_modules", target_is_directory=True)
        except OSError:
            pass
    argv = [
        str(esbuild), str(entry),
        "--bundle", "--format=esm", "--target=es2022", "--minify",
        f"--outfile={out}",
    ]
    env = {"PATH": "/usr/local/bin:/usr/bin:/bin",
           "NODE_PATH": str(REPO_ROOT / "node_modules")}
    proc = await asyncio.create_subprocess_exec(
        *argv, cwd=str(job.dir), env=env,
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120)
    except asyncio.CancelledError:
        if proc.returncode is None:
            proc.kill()
            try:
                await proc.wait()
            except ProcessLookupError:
                pass
        raise
    except asyncio.TimeoutError as exc:
        if proc.returncode is None:
            proc.kill()
            try:
                await proc.wait()
            except ProcessLookupError:
                pass
        raise PipelineError("esbuild timed out.", code="bundler_failed",
                            stage="bundling") from exc
    if proc.returncode != 0:
        raise PipelineError(
            f"esbuild failed: {stderr.decode('utf-8', 'replace')[:300]}",
            code="bundler_failed", stage="bundling")
    if not out.exists() or out.stat().st_size == 0:
        raise PipelineError("esbuild produced an empty bundle.",
                            code="bundler_failed", stage="bundling")


def _build_standalone(bundle_text: str, target_name: str) -> str:
    """Single self-contained HTML file with a base64-embedded ESM bundle."""
    encoded = base64.b64encode(bundle_text.encode("utf-8")).decode("ascii")
    title = re.sub(r"[^A-Za-z0-9 ]", "", target_name) or "Object"
    return STANDALONE_TEMPLATE.replace("__TITLE__", title).replace(
        "__BUNDLE_BASE64__", encoded)


STANDALONE_TEMPLATE = """<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>__TITLE__ — img2threejs standalone model</title>
<!--
  Generated by the img2threejs Hugging Face Space (https://huggingface.co/spaces/Mike0021/img2threejs)
  from a single reference image. Approximate procedural reconstruction;
  hidden sides are inferred, not measured. Built on hoainho/img2threejs (MIT).
  three.js is MIT.
-->
<style>
  html,body{margin:0;height:100%;background:#16181d;color:#e6e8ee;
    font:14px/1.5 system-ui,sans-serif}
  #viewer{position:fixed;inset:0}
  .tag{position:fixed;left:12px;bottom:10px;font-size:12px;color:#9aa0ad;
    background:rgba(22,24,29,.7);padding:4px 8px;border-radius:6px}
  #fallback{padding:2rem;max-width:60ch}
</style>
</head>
<body>
<div id="viewer" role="region" aria-label="3D model viewer"></div>
<div class="tag">__TITLE__ — procedural Three.js reconstruction (img2threejs). Drag to orbit, scroll to zoom.</div>
<noscript><div id="fallback">This model needs JavaScript + WebGL to render.</div></noscript>
<script type="module">
let url;
try {
  const bytes = Uint8Array.from(atob('__BUNDLE_BASE64__'), (char) => char.charCodeAt(0));
  url = URL.createObjectURL(new Blob([bytes], { type: 'text/javascript' }));
  const mod = await import(url);
  const el = document.getElementById('viewer');
  if (!document.createElement('canvas').getContext('webgl2') &&
      !document.createElement('canvas').getContext('webgl')) {
    throw new Error('WebGL is not available in this browser/GPU.');
  }
  mod.mountViewer(el, mod.makeModel,
    typeof mod.makeLights === 'function' ? mod.makeLights : null, {});
} catch (err) {
  const div = document.createElement('div');
  div.id = 'fallback';
  div.textContent = 'The generated model failed to render here: ' + (err && err.message || err);
  document.body.appendChild(div);
} finally {
  if (url) URL.revokeObjectURL(url);
}
</script>
</body>
</html>
"""