"""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 createModel 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 = """ __TITLE__ — img2threejs standalone model
__TITLE__ — procedural Three.js reconstruction (img2threejs). Drag to orbit, scroll to zoom.
"""