Fix audited generation, Wizard, World3D and review integrations

app/_launch_runtime.py

@@ -9947,6 +9947,9 @@ def director_pipeline_resume(pid: str):
 
 # ── Director Pipeline Dashboard ───────────────────────────────────────────
 
+from routers.director_review import create_director_review_router
+api.include_router(create_director_review_router(_workspace_dir))
+
 @api.get("/api/v1/director/pipelines")
 def list_saved_pipelines(limit: int = 0, offset: int = 0):
     """List saved pipeline states for the active workspace.
@@ -36924,13 +36927,14 @@ def _classic_redirect():
 from routers.scene_commands import create_scene_commands_router
 _scene_commands = SceneCommands(_workspace_dir)
 api.include_router(create_scene_commands_router(_scene_commands))
-from routers.world3d_export import create_world3d_export_router
+from routers.world3d_export import create_world3d_export_router, bind_world3d_renderer_origin
 from services.world3d_export import World3DExportService, command_catalog as world3d_export_catalog, command_handlers as world3d_export_handlers
 _world3d_export = World3DExportService(
     workspace_dir=_workspace_dir,
     registry_for=_task_registry,
     app_url=os.environ.get("HOCUS_APP_URL", ""),
 )
+bind_world3d_renderer_origin(api, _world3d_export)
 api.include_router(create_world3d_export_router(_world3d_export))
 
 from services.mcp_access import McpAccess
@@ -36948,7 +36952,7 @@ def _classic_redirect():
     command_receipt=_image_generation_commands.receipt,
     get_task=lambda workspace, task_id: _task_registry(workspace).get(task_id),
 )
-api.include_router(create_wizard_workflow_executor_router(_wizard_workflow_executor))
+api.include_router(create_wizard_workflow_executor_router(_wizard_workflow_executor, list_workspaces=_list_workspaces))
 api.include_router(create_wangp_mcp_router(
     token_getter=_mcp_access.token,
     handlers={"models": lambda args: get_model_options(args['model_type']) if args.get('model_type') else list_models(), "processors": wangp_capabilities, "status": get_status,

app/core_runtime.py

@@ -33,7 +33,7 @@
 from routers.system_capabilities import create_system_capabilities_router, require_capability_http
 from routers.user_diagnostics import create_user_diagnostics_router
 from routers.wizard_workflow_executor import create_wizard_workflow_executor_router
-from routers.world3d_export import create_world3d_export_router
+from routers.world3d_export import create_world3d_export_router, bind_world3d_renderer_origin
 from routers.workspace_collections import create_workspace_collections_router
 from services import (
     core_canonical_tasks,
@@ -110,11 +110,13 @@
     uploads_dir=core.uploads_dir,
     list_workspaces=core.list_workspaces,
 ))
-api.include_router(create_world3d_export_router(World3DExportService(
+_world3d_export = World3DExportService(
     workspace_dir=core.workspace_dir,
     registry_for=core_generation_commands.registry_for,
     app_url=os.environ.get("HOCUS_APP_URL", ""),
-)))
+)
+bind_world3d_renderer_origin(api, _world3d_export)
+api.include_router(create_world3d_export_router(_world3d_export))
 api.include_router(create_core_labs_router())
 api.include_router(create_series_assembly_router(
     resolve_workspace=labs._series_workspace,
@@ -143,7 +145,7 @@
     submit_command=_core_image_commands.submit,
     command_receipt=_core_image_commands.receipt,
     get_task=core_generation_commands.get_task,
-)))
+), list_workspaces=core.list_workspaces))
 api.include_router(create_character_kit_face_router(
     workspace_dir=core.workspace_dir,
     uploads_root=core.uploads_dir,

app/routers/director_review.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter, HTTPException, Request
+from starlette.concurrency import run_in_threadpool
+
+
+def create_director_review_router(workspace_dir) -> APIRouter:
+    router = APIRouter()
+
+    @router.put("/api/v1/director/pipelines/{pid}/review")
+    async def save(pid: str, request: Request):
+        from services.director_pipeline import PipelineBusyError
+        from services.director_review import save_review
+        try:
+            body = await request.json()
+            if not isinstance(body, dict) or not isinstance(body.get("workspace"), str):
+                raise ValueError("Use an explicit review workspace")
+            return await run_in_threadpool(save_review, workspace_dir(body["workspace"]), pid, body.get("commands"))
+        except PipelineBusyError as error:
+            raise HTTPException(409, str(error)) from error
+        except ValueError as error:
+            raise HTTPException(422, str(error)) from error
+
+    return router

app/routers/wizard_workflow_executor.py

@@ -11,11 +11,12 @@
 from services.image_generation_commands import command_error
 from services.wizard_workflow_executor import catalog, command_handlers
 from services.wizard_workflows import WizardWorkflowRevisionConflict
+from services.wizard_workflow_supervisor import workflow_lifespan
 
 
-def create_wizard_workflow_executor_router(executor) -> APIRouter:
+def create_wizard_workflow_executor_router(executor, *, list_workspaces=None, interval: float = 1.0) -> APIRouter:
     """Build the isolated executor router with an injected service."""
-    router = APIRouter()
+    router = APIRouter(lifespan=workflow_lifespan(executor, list_workspaces, interval) if list_workspaces else None)
 
     def _translate(error: Exception) -> HTTPException:
         if isinstance(error, HTTPException):

app/routers/world3d_export.py

@@ -16,7 +16,6 @@
     RECEIPT_OPERATION,
     command_catalog,
     command_handlers,
-    export_capabilities,
     http_error,
 )
 
@@ -30,7 +29,7 @@ def commands():
 
     @router.get("/api/v1/scenes/world3d/export/capabilities")
     def capabilities():
-        return export_capabilities()
+        return service.capabilities()
 
     @router.post("/api/v1/scenes/world3d/export")
     async def submit(request: Request):
@@ -60,6 +59,15 @@ async def cancel(request: Request):
     return router
 
 
+def bind_world3d_renderer_origin(api, service) -> None:
+    @api.middleware("http")
+    async def bind_renderer_origin(request: Request, call_next):
+        # Use the listening socket, never a client-controlled Host header.
+        server = request.scope.get("server")
+        if not service.app_url and server:
+            service.app_url = f"http://127.0.0.1:{server[1]}"
+        return await call_next(request)
+
 __all__ = [
     "CANCEL_OPERATION",
     "OPERATION",

app/services/core_generation_commands.py

@@ -33,6 +33,8 @@
 
 _REGISTRIES: dict[str, TaskRegistry] = {}
 _LOCK = threading.Lock()
+_DISPATCH_OWNER = f"core-{uuid.uuid4().hex}"
+_DISPATCH_LOCK = threading.Lock()
 
 
 def registry_for(workspace: str) -> TaskRegistry:
@@ -211,7 +213,7 @@ def receipt(self, workspace: str, intent_id: str) -> dict[str, Any]:
         except (OSError, sqlite3.Error) as error:
             raise command_error(503, "storage_unavailable", "Command storage is unavailable") from error
 
-    def _ensure_job(self, workspace: str, job_id: str, params: dict[str, Any]) -> None:
+    def _ensure_job(self, workspace: str, job_id: str, params: dict[str, Any], task_id: str) -> None:
         if core_remote_image.get_job(job_id) is not None:
             return
         core_remote_image.start_job(
@@ -226,6 +228,7 @@ def _ensure_job(self, workspace: str, job_id: str, params: dict[str, Any]) -> No
             },
             workspace=workspace,
             job_id=job_id,
+            on_update=lambda: get_task(workspace, task_id),
         )
 
     async def submit(self, command, *, trusted_tool=None, submission_context=None):
@@ -249,7 +252,19 @@ async def submit(self, command, *, trusted_tool=None, submission_context=None):
                 task_fields=_task_fields(workspace, job_id, params),
                 fingerprint_version=frozen["fingerprint_version"],
             )
-            self._ensure_job(workspace, admitted["receipt"]["result"]["job_id"], params)
+            result = admitted["receipt"]["result"]
+            # A concurrent replay must not mistake claim → job creation for a
+            # previous process losing its provider outcome.
+            with _DISPATCH_LOCK:
+                if registry.claim_command_dispatch(command["intent_id"], _DISPATCH_OWNER):
+                    self._ensure_job(workspace, result["job_id"], params, result["task_id"])
+                elif core_remote_image.get_job(result["job_id"]) is None:
+                    task = registry.get(result["task_id"])
+                    if task and task["status"] in ACTIVE_STATUSES:
+                        task = registry.update(task["id"], status="interrupted", force=True,
+                                               message="Provider outcome unknown; create a new attempt to retry")
+                    if task:
+                        core_remote_image.restore_job(task)
             return admitted
         except ImageGenerationSpecError as error:
             raise command_error(422, "invalid_command", str(error)) from error

app/services/core_remote_image.py

@@ -133,14 +133,35 @@ def cancel_job(job_id: str) -> dict[str, Any] | None:
             job.update(status="cancelled", phase="cancelled", message="Cancelled", finished_at=time.time())
         else:
             job.update(status="cancelling", phase="cancelling", message="Cancellation requested")
-        return _public(job)
+        result = _public(job)
+        notify = job.get("_on_update")
+    if notify:
+        notify()
+    return result
 
 
 def _patch(job_id: str, **fields: Any) -> None:
     with _LOCK:
         job = _JOBS.get(job_id)
         if job:
             job.update(fields)
+        notify = job.get("_on_update") if job else None
+    if notify:
+        notify()
+
+
+def restore_job(task: dict[str, Any]) -> None:
+    """Restore polling from a durable task without invoking a provider."""
+    job_id = task["backend_job_id"]
+    with _LOCK:
+        _JOBS.setdefault(job_id, {
+            "id": job_id, "task_id": task["id"], "root_task_id": task.get("root_id"),
+            "status": task["status"], "phase": task["status"],
+            "progress": 100 if task["status"] == "completed" else 0,
+            "message": task.get("message") or "", "error": task.get("error"),
+            "output_files": list(task.get("result_refs") or []),
+            "created_at": task.get("created_at"), "workspace": task.get("workspace"),
+        })
 
 
 def encode_subject_reference(source: str, workspace: str) -> str:
@@ -164,7 +185,7 @@ def encode_subject_reference(source: str, workspace: str) -> str:
     return local_image_data_uri(path)
 
 
-def start_job(body: dict[str, Any], *, workspace: str, job_id: str | None = None) -> dict[str, Any]:
+def start_job(body: dict[str, Any], *, workspace: str, job_id: str | None = None, on_update=None) -> dict[str, Any]:
     from services import execution_mode
 
     prompt = prepare_prompt(str(body.get("prompt") or ""))
@@ -186,6 +207,7 @@ def start_job(body: dict[str, Any], *, workspace: str, job_id: str | None = None
             "output_files": [], "error": None, "workspace": workspace,
             "created_at": now, "started_at": None, "finished_at": None,
             "_cancel_requested": False,
+            "_on_update": on_update,
             "request": {"prompt": prompt, "aspect_ratio": ratio, "subject_reference": subject},
         }
         _JOBS[job_id] = job
@@ -203,9 +225,10 @@ def _run(job_id: str) -> None:
         _patch(job_id, status="running", phase="running", progress=10, started_at=time.time(),
                message="Calling MiniMax Image-01")
         with _LOCK:
-            if (_JOBS.get(job_id) or {}).get("_cancel_requested"):
-                _patch(job_id, status="cancelled", phase="cancelled", message="Cancelled", finished_at=time.time())
-                return
+            cancelled = (_JOBS.get(job_id) or {}).get("_cancel_requested")
+        if cancelled:
+            _patch(job_id, status="cancelled", phase="cancelled", message="Cancelled", finished_at=time.time())
+            return
         result = generate_image(
             api_key=resolve_minimax_key(core.services_raw(), "image"),
             prompt=str(request.get("prompt") or ""),

app/services/director_review.py

@@ -0,0 +1,75 @@
+"""Persist review decisions onto the existing Director pipeline, atomically."""
+from pathlib import Path
+import json
+import time
+
+from services.director_pipeline import (
+    _exclusive_pipeline_operation, _find_pipeline_file, _pipeline_file_lock,
+    _write_pipeline_json_unlocked,
+)
+
+
+def _select_take(clip: dict, name: str, workspace: Path, state: dict) -> None:
+    attempts = {item.get("filename"): item for item in clip.get("video_attempts", []) if isinstance(item, dict)}
+    available = set(attempts) | {clip.get("video_filename"), clip.get("selected_video_filename")}
+    if not isinstance(name, str) or not name or name not in available or Path(name).name != name:
+        raise ValueError("Select an existing take from this shot")
+    file = workspace / name
+    if not file.is_file() or file.resolve().parent != workspace.resolve() or file.stat().st_size == 0:
+        raise ValueError("The selected take is missing from this workspace")
+    changed = name != (clip.get("selected_video_filename") or clip.get("video_filename"))
+    clip.update(selected_video_filename=name, video_filename=name)
+    if changed:
+        clip.update(video_stale=False, tag=None)
+        segments = clip.get("h3_segments") or []
+        if len(segments) == 1:
+            attempt = attempts.get(name, {})
+            segment = segments[0]
+            segment.update(filename=name, prompt=attempt.get("prompt") or segment.get("prompt", ""),
+                           seed=attempt.get("seed", segment.get("seed")), stale=False, updated_at=time.time())
+            if attempt.get("video_length"):
+                segment["frames"] = attempt["video_length"]
+    if name not in state.get("output_files", []):
+        state.setdefault("output_files", []).append(name)
+
+
+def _apply_review(state: dict, commands: list, workspace: Path, pid: str) -> None:
+    clips = {clip.get("index", index): clip for index, clip in enumerate(state.get("clips", []))}
+    for command in commands:
+        if not isinstance(command, dict) or command.get("pipelineId") != pid:
+            raise ValueError("Review command belongs to another production")
+        index = command.get("clipIndex")
+        if type(index) is not int or index not in clips:
+            raise ValueError("Review shot was not found")
+        clip = clips[index]
+        kind = command.get("type")
+        if kind == "select_take":
+            _select_take(clip, command.get("filename"), workspace, state)
+        elif kind == "tag_clip":
+            tag = command.get("tag")
+            if tag not in (None, "good", "needs_work"):
+                raise ValueError("Invalid review decision")
+            if tag == "good" and (not clip.get("video_filename") or clip.get("video_stale")):
+                raise ValueError("Only a completed current take can be approved")
+            clip["tag"] = tag
+        elif kind == "note_clip":
+            notes = command.get("notes")
+            if not isinstance(notes, str) or len(notes) > 8000:
+                raise ValueError("Review notes must contain at most 8000 characters")
+            clip["review_notes"] = notes
+        else:
+            raise ValueError("Unsupported review command")
+
+
+@_exclusive_pipeline_operation
+def save_review(workspace: str, pid: str, commands: list) -> dict:
+    if not isinstance(commands, list) or len(commands) > 1500:
+        raise ValueError("Use a bounded list of review decisions")
+    path = _find_pipeline_file(workspace, pid)
+    if not path or Path(path).resolve().parent != Path(workspace).resolve():
+        raise ValueError("Production not found in this workspace")
+    with _pipeline_file_lock:
+        state = json.loads(Path(path).read_text(encoding="utf-8"))
+        _apply_review(state, commands, Path(workspace), pid)
+        _write_pipeline_json_unlocked(path, state)
+    return state