feat(workflows): support scene model artifacts

README.md

@@ -153,6 +153,18 @@ original behavior.
 ## Workflows
 Start with a basic workflow first. For example, on the "Workflows" tab, try: Image -> Generate Mesh -> Add to Scene. Make sure there is a connection between each of the steps. Go to the "Generate" tab, make sure the workflow is selected, then click on "Generate 3D Model". Click on "Settings/Logs/Errors" to see any issues.
 
+Model extensions may also declare `scene` as a node input or output. A scene is
+a workspace directory containing `scene-manifest.json` with schema
+`modly.scene-manifest.v1`; it is not an arbitrary JSON file. Use the **Load
+Scene** workflow node to select and validate an existing scene directory.
+Scene-capable generators implement `generate_artifact(input_kind,
+artifact_path, ...)`; legacy image generators and `POST /generate/from-image`
+remain unchanged. The generic `POST /generate/from-artifact` boundary currently
+accepts only `scene`, leaving future artifact kinds to separate reviewed changes.
+For this first contract, `scene` is model-only and must be declared as the single
+`input` value (not inside `inputs`); process and mixed-input scene nodes are rejected.
+Model nodes may still accept multiple images and produce a scene.
+
 
 ## Modly CLI
 

api/README.md

@@ -29,6 +29,7 @@ uvicorn main:app --host 127.0.0.1 --port 8765 --reload
 | GET | `/model/status` | Model download / load status |
 | GET | `/model/download` | SSE stream of download progress |
 | POST | `/generate/from-image` | Start image-to-3D job |
+| POST | `/generate/from-artifact` | Start a typed-artifact model job (`scene` only) |
 | GET | `/generate/status/{job_id}` | Poll job status |
 
 ## Model

api/routers/extensions.py

@@ -23,7 +23,8 @@ async def reload_extensions(payload: dict | None = Body(default=None)):
         candidate = payload.get("validationCapability")
         if isinstance(candidate, dict):
             validation_capability = candidate
-    generator_registry.reload(validation_capability)
+    # Off the event loop: reload waits for any in-progress model load.
+    await asyncio.to_thread(generator_registry.reload, validation_capability)
     return {
         "reloaded": True,
         "models":   list(generator_registry._generators.keys()),

api/routers/generation.py

@@ -4,7 +4,9 @@
 import time
 import traceback
 import uuid
-from typing import Dict
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from typing import Dict, Optional, Union
 from fastapi import APIRouter, File, Form, UploadFile, HTTPException, BackgroundTasks
 from services.generators.base import smooth_progress, GenerationCancelled
 
@@ -14,7 +16,12 @@
 # binding captured at import would keep writing output to the old directory.
 import services.generator_registry as registry
 from services.generator_registry import generator_registry
-from schemas.generation import JobStatus
+from schemas.generation import GenerateFromArtifactRequest, JobStatus
+from services.artifact_input import (
+    RESERVED_ARTIFACT_PARAMS,
+    TypedArtifactInput,
+    validate_artifact_input,
+)
 
 router = APIRouter(tags=["generation"])
 
@@ -26,6 +33,14 @@
 _cancelled: set = set()
 _cancel_events: Dict[str, threading.Event] = {}
 _completed_at: Dict[str, float] = {}
+_job_generators: Dict[str, object] = {}
+# A pinned generation owns the complete switch/load/generate lifecycle.  Keeping
+# that lifecycle on one dedicated worker provides process-wide serialization
+# without parking default-executor workers on a blocking lock acquisition.
+_pinned_generation_executor = ThreadPoolExecutor(
+    max_workers=1,
+    thread_name_prefix="modly-pinned-generation",
+)
 
 _JOB_TTL = 1800  # purge terminal jobs after 30 minutes
 
@@ -37,6 +52,7 @@ def _purge_old_jobs() -> None:
         _jobs.pop(jid, None)
         _cancelled.discard(jid)
         _cancel_events.pop(jid, None)
+        _job_generators.pop(jid, None)
         _completed_at.pop(jid, None)
 
 
@@ -105,11 +121,10 @@ async def generate_from_image(
     # Verify the requested model exists in the registry
     try:
         generator_registry.get_generator(model_id)
+        output_kind = generator_registry.get_manifest(model_id).get("output", "mesh")
     except ValueError as e:
         raise HTTPException(400, str(e))
 
-    generator_registry.switch_model(model_id)
-
     # Parse model-specific params from JSON and merge with common fields
     try:
         model_params = json.loads(params)
@@ -131,8 +146,41 @@ async def generate_from_image(
     _jobs[job_id] = job
     _cancel_events[job_id] = threading.Event()
 
-    background_tasks.add_task(_run_generation, job_id, image_bytes, full_params, collection)
+    background_tasks.add_task(
+        _run_generation, job_id, image_bytes, full_params, collection, output_kind, model_id
+    )
+
+    return {"job_id": job_id}
 
+
+@router.post("/from-artifact")
+async def generate_from_artifact(
+    request: GenerateFromArtifactRequest,
+    background_tasks: BackgroundTasks,
+):
+    """Queue a validated typed artifact without serializing it as image bytes."""
+    try:
+        manifest = generator_registry.get_manifest(request.model_id)
+    except (KeyError, ValueError) as exc:
+        raise HTTPException(400, str(exc)) from exc
+    declared = manifest.get("inputs") or [manifest.get("input", "image")]
+    if request.input_kind not in declared:
+        raise HTTPException(400, f"Model {request.model_id} does not accept {request.input_kind} input")
+    try:
+        artifact = validate_artifact_input(registry.WORKSPACE_DIR, request.input_kind, request.input_path)
+    except ValueError as exc:
+        raise HTTPException(400, str(exc)) from exc
+
+    params = {k: v for k, v in request.params.items() if k not in RESERVED_ARTIFACT_PARAMS}
+    params["scene_manifest_path"] = str(artifact.path)
+    collection = sanitize_collection(request.collection)
+    job_id = str(uuid.uuid4())
+    _purge_old_jobs()
+    _jobs[job_id] = JobStatus(job_id=job_id, status="pending", progress=0)
+    _cancel_events[job_id] = threading.Event()
+    background_tasks.add_task(
+        _run_generation, job_id, artifact, params, collection, manifest.get("output", "mesh"), request.model_id
+    )
     return {"job_id": job_id}
 
 
@@ -156,11 +204,10 @@ async def cancel_job(job_id: str):
     if job.status in ("pending", "running"):
         job.status = "cancelled"
         _completed_at[job_id] = time.monotonic()
-    # Kill the active generator subprocess immediately so inference stops now.
-    # _run_generation will catch the resulting exception, see job_id in _cancelled,
-    # and return cleanly without setting an error status.
+    # Kill only the subprocess bound to this job. A queued cancellation must not
+    # terminate whichever earlier job currently owns the active generator.
     try:
-        gen = generator_registry._generators.get(generator_registry._active_id)
+        gen = _job_generators.get(job_id)
         if gen is not None and hasattr(gen, "_proc") and gen._proc and gen._proc.poll() is None:
             gen._proc.kill()
             gen._loaded = False
@@ -170,9 +217,66 @@ async def cancel_job(job_id: str):
     return {"cancelled": True}
 
 
-async def _run_generation(job_id: str, image_bytes: bytes, params: dict, collection: str = "Default") -> None:
+async def _run_generation(
+    job_id: str,
+    model_input: Union[bytes, TypedArtifactInput],
+    params: dict,
+    collection: str = "Default",
+    output_kind: str = "mesh",
+    model_id: Optional[str] = None,
+) -> None:
+    # Pinned jobs share one model lifecycle. Switching is deliberately deferred
+    # until this job runs on the dedicated worker: request-time switches can
+    # otherwise unload a running job or leave A loading beside B.  Crucially,
+    # queued jobs are executor work items rather than default-executor threads
+    # blocked on a lock, so cancelling a waiter cannot orphan queue ownership or
+    # starve the worker that performs generation.
+    loop = asyncio.get_running_loop()
+    executor = _pinned_generation_executor if model_id is not None else None
+    # Shown while this job waits behind another one on the single worker;
+    # _run_generation_impl clears it as soon as the job actually starts.
+    queued_job = _jobs.get(job_id)
+    if queued_job is not None and executor is not None:
+        queued_job.step = "Waiting for the previous generation…"
+    future = loop.run_in_executor(
+        executor,
+        _run_generation_impl,
+        job_id,
+        model_input,
+        params,
+        collection,
+        output_kind,
+        model_id,
+    )
+    try:
+        await future
+    except asyncio.CancelledError:
+        # asyncio cancellation attempts to cancel a queued concurrent future.
+        # If it has already begun, the event lets the generator stop safely.
+        _cancelled.add(job_id)
+        cancel_event = _cancel_events.get(job_id)
+        if cancel_event is not None:
+            cancel_event.set()
+        job = _jobs.get(job_id)
+        if job is not None and job.status in ("pending", "running"):
+            job.status = "cancelled"
+            _completed_at[job_id] = time.monotonic()
+        raise
+
+
+def _run_generation_impl(
+    job_id: str,
+    model_input: Union[bytes, TypedArtifactInput],
+    params: dict,
+    collection: str,
+    output_kind: str,
+    model_id: Optional[str],
+) -> None:
+    if job_id in _cancelled:
+        return
     job = _jobs[job_id]
     job.status = "running"
+    job.step = None
 
     def progress_cb(pct: int, step: str = "") -> None:
         # Monotonic: the loading phase walks the bar up on a background thread and
@@ -184,13 +288,17 @@ def progress_cb(pct: int, step: str = "") -> None:
             job.step = step
 
     try:
-        loop = asyncio.get_running_loop()
-
-        # Check if the model needs to be loaded BEFORE calling get_active(),
-        # because get_active() loads the model in a blocking manner.
-        # active_status() is an instantaneous operation (simple dict lookup).
-        if not generator_registry.active_status()["loaded"]:
-            active = generator_registry.active_status()
+        # Check if the model needs to be loaded BEFORE calling the generator
+        # getter, because that call can load the model in a blocking manner.
+        get_generator = (lambda: generator_registry.activate_ready_generator(model_id)) \
+            if model_id is not None else generator_registry.get_active
+        if model_id is not None:
+            _job_generators[job_id] = generator_registry.get_generator(model_id)
+        status_reader = (lambda: generator_registry.model_status(model_id)) \
+            if model_id is not None else generator_registry.active_status
+        status = status_reader()
+        if not status["loaded"]:
+            active = status
             model_name = active['name']
             init_label = f"Downloading {model_name}…" if not active['downloaded'] else f"Loading {model_name}…"
             progress_cb(0, init_label)
@@ -202,12 +310,13 @@ def progress_cb(pct: int, step: str = "") -> None:
             )
             load_thread.start()
             try:
-                gen = await loop.run_in_executor(None, generator_registry.get_active)
+                gen = get_generator()
             finally:
                 stop_load_evt.set()
         else:
-            gen = await loop.run_in_executor(None, generator_registry.get_active)
+            gen = get_generator()
 
+        _job_generators[job_id] = gen
         if job_id in _cancelled:
             return
 
@@ -217,18 +326,39 @@ def progress_cb(pct: int, step: str = "") -> None:
         gen.outputs_dir = coll_dir
 
         cancel_event = _cancel_events.get(job_id)
-        import inspect
-        supports_cancel = "cancel_event" in inspect.signature(gen.generate).parameters
-        output_path = await loop.run_in_executor(
-            None,
-            lambda: gen.generate(image_bytes, params, progress_cb, cancel_event)
-                    if supports_cancel
-                    else gen.generate(image_bytes, params, progress_cb),
-        )
+        if isinstance(model_input, TypedArtifactInput):
+            # Revalidate just before crossing the inference boundary. The
+            # subprocess runner repeats this check inside the worker.
+            from services.artifact_input import revalidate_artifact_input
+            model_input = revalidate_artifact_input(registry.WORKSPACE_DIR, model_input)
+            import inspect
+            supports_cancel = "cancel_event" in inspect.signature(gen.generate_artifact).parameters
+            output_path = (
+                gen.generate_artifact(model_input.kind, model_input.path, params, progress_cb, cancel_event)
+                if supports_cancel
+                else gen.generate_artifact(model_input.kind, model_input.path, params, progress_cb)
+            )
+        else:
+            import inspect
+            supports_cancel = "cancel_event" in inspect.signature(gen.generate).parameters
+            output_path = (
+                gen.generate(model_input, params, progress_cb, cancel_event)
+                if supports_cancel
+                else gen.generate(model_input, params, progress_cb)
+            )
 
         if job_id in _cancelled:
             return
 
+        output_path = Path(output_path).resolve(strict=True)
+        if output_kind == "scene":
+            from services.scene_input import validate_scene_input
+            try:
+                output_relative = output_path.relative_to(registry.WORKSPACE_DIR.resolve())
+                output_path = validate_scene_input(registry.WORKSPACE_DIR, output_relative.as_posix())
+            except (OSError, ValueError) as exc:
+                raise ValueError("Generated scene output is not a valid workspace scene") from exc
+
         job.status   = "done"
         job.progress = 100
         _completed_at[job_id] = time.monotonic()
@@ -253,3 +383,5 @@ def progress_cb(pct: int, step: str = "") -> None:
         job.status = "error"
         job.error  = tb.strip()
         _completed_at[job_id] = time.monotonic()
+    finally:
+        _job_generators.pop(job_id, None)

api/routers/model.py

@@ -59,20 +59,20 @@ def _check_download_control(control: dict[str, threading.Event]) -> None:
 @router.get("/status")
 async def model_status():
     """Status of the active model."""
-    return generator_registry.active_status()
+    return await asyncio.to_thread(generator_registry.active_status)
 
 
 @router.get("/all")
 async def all_models_status():
     """Status of all known models (downloaded, loaded, required VRAM)."""
-    return generator_registry.all_status()
+    return await asyncio.to_thread(generator_registry.all_status)
 
 
 @router.get("/params")
 async def model_params(model_id: Optional[str] = None):
     """Parameter schema of the active model (or a specified model)."""
     try:
-        return generator_registry.params_schema(model_id)
+        return await asyncio.to_thread(generator_registry.params_schema, model_id)
     except KeyError:
         raise HTTPException(404, f"Unknown model ID: {model_id}")
 
@@ -81,7 +81,7 @@ async def model_params(model_id: Optional[str] = None):
 async def switch_model(model_id: str):
     """Switch the active model."""
     try:
-        generator_registry.switch_model(model_id)
+        await asyncio.to_thread(generator_registry.switch_model, model_id)
         return {"active": model_id}
     except ValueError as e:
         raise HTTPException(400, str(e))
@@ -90,7 +90,8 @@ async def switch_model(model_id: str):
 @router.post("/unload-all")
 async def unload_all_models():
     """Unloads all models from memory to free VRAM/RAM."""
-    generator_registry.unload_all()
+    # Off the event loop: unloading waits for any in-progress model load.
+    await asyncio.to_thread(generator_registry.unload_all)
     # Force Python to release memory back to the OS
     import gc
     gc.collect()

api/routers/settings.py

@@ -1,3 +1,4 @@
+import asyncio
 import os
 from fastapi import APIRouter
 from pydantic import BaseModel
@@ -28,9 +29,11 @@ async def get_paths():
 
 @router.post("/paths")
 async def update_paths(body: PathsUpdate):
-    reg_module.generator_registry.update_paths(
-        models_dir    = Path(body.models_dir)    if body.models_dir    else None,
-        workspace_dir = Path(body.workspace_dir) if body.workspace_dir else None,
+    # Off the event loop: changing paths waits for any in-progress model load.
+    await asyncio.to_thread(
+        reg_module.generator_registry.update_paths,
+        Path(body.models_dir)    if body.models_dir    else None,
+        Path(body.workspace_dir) if body.workspace_dir else None,
     )
     return {
         "models_dir":    str(reg_module.MODELS_DIR),