feat(provenance): trace Studio generations from Wizard to asset

app/_launch_runtime.py

@@ -449,7 +449,7 @@ async def trace_user_mutations(request: Request, call_next):
     unregister_abort_state,
     update_job,
 )
-from services.asset_manifest import publish_generation_sidecar
+from services.asset_manifest import publish_generation_sidecar, publish_generation_sidecar_best_effort
 from services import resource_scheduler
 
 _jobs: dict = {}
@@ -552,6 +552,7 @@ def _persist_generation_job(job: dict) -> None:
             "created_at": job.get("created_at", time.time()),
             "params": copy.deepcopy(job.get("params") or {}),
             "workspace": job.get("workspace") or "default",
+            "provenance": copy.deepcopy(job.get("provenance") or {}),
         })
     except Exception as exc:
         # Persistence should protect a generation, never prevent it from
@@ -576,6 +577,7 @@ def _new_generation_job(
     created_at: float | None = None,
     recovered: bool = False,
     reserve_generation: bool = True,
+    provenance: dict | None = None,
 ) -> dict:
     frozen_params = copy.deepcopy(params)
     execution_mode.validate_generation(workspace)
@@ -610,8 +612,19 @@ def _new_generation_job(
                 f"{len(timeline['intervals'])} timed interval(s) across "
                 f"{timeline['duration_seconds']:.3f}s."
             )
+    resolved_job_id = job_id or uuid.uuid4().hex[:8]
+    canonical_task_id = f"task-generation-{resolved_job_id}"
+    owner_id = str(frozen_params.get("_director_pipeline_id") or "")
+    if owner_id.startswith("series:"):
+        canonical_root_task_id = f"task-series-render-{owner_id.split(':', 1)[1]}"
+    elif owner_id:
+        canonical_root_task_id = f"task-director-{owner_id}"
+    else:
+        canonical_root_task_id = canonical_task_id
     job = {
-        "id": job_id or uuid.uuid4().hex[:8],
+        "id": resolved_job_id,
+        "task_id": canonical_task_id,
+        "root_task_id": canonical_root_task_id,
         "status": "queued",
         "progress": 0,
         "step": 0,
@@ -627,6 +640,7 @@ def _new_generation_job(
         "error": None,
         "workspace": workspace,
         "out_dir": _workspace_dir(workspace),
+        "provenance": copy.deepcopy(provenance or {}),
         "recovered": recovered,
     }
     # Reserve FIFO order synchronously. Starting one thread per request is
@@ -639,13 +653,54 @@ def _new_generation_job(
         try:
             task = publisher(job)
             if isinstance(task, dict):
-                job["task_id"] = task.get("id")
-                job["root_task_id"] = task.get("root_id")
+                job["task_id"] = task.get("id") or job["task_id"]
+                job["root_task_id"] = (
+                    task.get("root_id") or task.get("id") or job["root_task_id"]
+                )
         except Exception as exc:
             print(f"[Task registry] Could not publish generation {job['id']}: {exc}")
     return job
 
 
+def _publish_generation_sidecar_for_studio_job(
+    job: dict,
+    output_path: str,
+    sidecar: dict,
+    *,
+    tool: str = "studio",
+) -> None:
+    """Publish one Studio result with its initiating command and real location."""
+    provenance = job.get("provenance") if isinstance(job.get("provenance"), dict) else {}
+    command = provenance.get("command") if isinstance(provenance.get("command"), dict) else {}
+    payload = dict(sidecar)
+    params = dict(payload.get("params") or {})
+    model_type = str(params.get("model_type") or "")
+    params.setdefault("provider", "minimax" if model_type.startswith("minimax:") else "local")
+    payload["params"] = params
+    if not payload.get("job_id"):
+        payload["job_id"] = job.get("id")
+    if not payload.get("task_id"):
+        payload["task_id"] = job.get("task_id")
+    if not payload.get("root_task_id"):
+        payload["root_task_id"] = job.get("root_task_id") or job.get("task_id")
+    payload["created_at"] = job.get("created_at") or payload.get("created_at") or time.time()
+    payload["queued_at"] = job.get("created_at") or payload.get("queued_at")
+    payload["started_at"] = job.get("started_at") or payload.get("started_at")
+    payload["completed_at"] = job.get("finished_at") or time.time()
+    for key in ("command_id", "workflow_id", "run_id"):
+        if command.get(key):
+            payload.setdefault(key, command[key])
+    publish_generation_sidecar_best_effort(
+        output_path,
+        payload,
+        workspace_id=provenance.get("workspace_id"),
+        output_folder=job.get("workspace"),
+        tool=tool,
+        actor=provenance.get("actor"),
+        capability=provenance.get("capability"),
+    )
+
+
 def _register_manual_generation_job(job: dict) -> dict:
     """Publish older edit/tool jobs before their background thread starts."""
     job_id = str(job.get("id") or "")
@@ -7439,7 +7494,13 @@ def model3d_capabilities():
 @api.post("/api/v1/model3d/generate")
 async def generate_model3d(request: Request):
     from services import model3d_service
+    from services.generation_provenance import normalize_submission_provenance
+
     body = await request.json()
+    body["provenance"] = normalize_submission_provenance(body.pop("provenance", None))
+    collection_id = body["provenance"].get("workspace_id")
+    if collection_id and not _workspace_collection_registry.get(collection_id):
+        raise HTTPException(status_code=400, detail="Unknown Workspace collection")
     workspace = body.get("workspace") if "workspace" in body else _get_active_workspace()
     _workspace_dir(workspace)
     try:
@@ -11150,7 +11211,13 @@ def _plan_windows():
 @api.post("/api/v1/generate")
 async def generate(request: Request):
     """Submit a generation job. Returns immediately with a job_id."""
+    from services.generation_provenance import normalize_submission_provenance
+
     body = await request.json()
+    provenance = normalize_submission_provenance(body.pop("provenance", None))
+    collection_id = provenance.get("workspace_id")
+    if collection_id and not _workspace_collection_registry.get(collection_id):
+        raise HTTPException(status_code=400, detail="Unknown Workspace collection")
     # Execution mode is a boot-time trust boundary, never a request option.
     # Discard spoofed private fields before validating the captured workspace.
     body.pop("_execution_mode", None)
@@ -11619,6 +11686,7 @@ async def generate(request: Request):
         body,
         workspace,
         reserve_generation=not h3_preplan_pending,
+        provenance=provenance,
     )
     job_id = job["id"]
     job["out_dir"] = job_out_dir
@@ -22842,15 +22910,9 @@ def _run_sfx_generation(job: dict, raw_params: dict, start_time: float):
                 "generation_time": round(elapsed),
                 "created_at": time.time(),
             }
-            try:
-                publish_generation_sidecar(
-                    os.path.join(out_dir, fname),
-                    sidecar,
-                    workspace_id=job.get("workspace"),
-                    tool="studio-sfx",
-                )
-            except Exception:
-                pass
+            _publish_generation_sidecar_for_studio_job(
+                job, os.path.join(out_dir, fname), sidecar,
+            )
 
         completed = finish_job(
             job,
@@ -23647,12 +23709,7 @@ def publish_progress(message: str, value: int, step: int, total: int) -> None:
         "simulated": True,
         "execution_mode": "simulate",
     }
-    publish_generation_sidecar(
-        generated_path,
-        sidecar,
-        workspace_id=job.get("workspace"),
-        tool="studio",
-    )
+    _publish_generation_sidecar_for_studio_job(job, generated_path, sidecar)
     if not finalize:
         return update_job(
             job,
@@ -23852,12 +23909,7 @@ def _legacy_h3_progress(
                 for path in generated:
                     file_sidecar = dict(sidecar)
                     file_sidecar["output_filename"] = os.path.basename(path)
-                    publish_generation_sidecar(
-                        path,
-                        file_sidecar,
-                        workspace_id=job.get("workspace"),
-                        tool="studio-h3-legacy",
-                    )
+                    _publish_generation_sidecar_for_studio_job(job, path, file_sidecar)
 
                 if not finalize:
                     return update_job(
@@ -24484,15 +24536,9 @@ def _write_output_sidecars(file_names):
                     else:
                         file_sidecar.pop("director_clip_index", None)
                     file_sidecar["output_filename"] = fname
-                    try:
-                        publish_generation_sidecar(
-                            os.path.join(out_dir, fname),
-                            file_sidecar,
-                            workspace_id=job.get("workspace"),
-                            tool="studio",
-                        )
-                    except Exception:
-                        pass
+                    _publish_generation_sidecar_for_studio_job(
+                        job, os.path.join(out_dir, fname), file_sidecar,
+                    )
 
             is_multiclip = total_tasks > 1 and any(t.get('params', {}).get('multi_clip_info') for t in queue)
 
@@ -26403,6 +26449,7 @@ def resume_generation_queue():
                 reserve_generation=not isinstance(
                     params.get("_h3_window_plan_pending"), dict,
                 ),
+                provenance=record.get("provenance") if isinstance(record.get("provenance"), dict) else None,
             )
             _jobs[job_id] = job
             _persist_generation_job(job)
@@ -36154,6 +36201,8 @@ def _publish_generation_task(job: dict) -> dict:
     task_id = f"task-generation-{legacy_id}"
     params = job.get("params") if isinstance(job.get("params"), dict) else {}
     owner_id = str(params.get("_director_pipeline_id") or "")
+    provenance = job.get("provenance") if isinstance(job.get("provenance"), dict) else {}
+    command = provenance.get("command") if isinstance(provenance.get("command"), dict) else {}
     if owner_id.startswith("series:"):
         series_job_id = owner_id.split(":", 1)[1]
         parent_task_id = f"task-series-render-{series_job_id}"
@@ -36215,6 +36264,12 @@ def _publish_generation_task(job: dict) -> dict:
         metadata={
             "adapter": "generation", "generation_details": details,
             "owner_pipeline_id": owner_id,
+            "actor": provenance.get("actor") or "unknown",
+            "tool": provenance.get("tool") or "studio",
+            "capability": provenance.get("capability"),
+            "command_id": command.get("command_id"),
+            "workflow_id": command.get("workflow_id"),
+            "run_id": command.get("run_id"),
         },
     )
 
@@ -36364,6 +36419,8 @@ def _publish_generic_legacy_task(record: dict, adapter: str) -> dict | None:
         or ""
     ) or None
     request_body = record.get("request") if isinstance(record.get("request"), dict) else {}
+    provenance = record.get("provenance") if isinstance(record.get("provenance"), dict) else {}
+    command = provenance.get("command") if isinstance(provenance.get("command"), dict) else {}
     provider = str(
         record.get("provider")
         or request_body.get("writingProvider")
@@ -36425,11 +36482,20 @@ def _publish_generic_legacy_task(record: dict, adapter: str) -> dict | None:
                     and adapter != "comic-plan"),
         resumable=resumable, recoverable=resumable,
         error=({"message": str(record.get("error")), "retryable": resumable} if record.get("error") else None),
-        result_refs=list(record.get("output_files") or ([record["output"]] if record.get("output") else [])),
+        result_refs=list(record.get("output_files") or (
+            [record["output"]] if record.get("output") else
+            [record["filename"]] if record.get("filename") else []
+        )),
         metadata={
             "adapter": adapter,
             "cancel_mode": record.get("cancel_mode"),
             "safe_boundary": record.get("safe_boundary"),
+            "actor": provenance.get("actor") or "unknown",
+            "tool": provenance.get("tool") or adapter,
+            "capability": provenance.get("capability"),
+            "command_id": command.get("command_id"),
+            "workflow_id": command.get("workflow_id"),
+            "run_id": command.get("run_id"),
         },
     )
 

app/services/asset_manifest.py

@@ -395,6 +395,7 @@ def adapt_legacy_sidecar(
         parameters=params,
         timing={
             "created_at": legacy.get("created_at"),
+            "queued_at": legacy.get("queued_at"),
             "started_at": started_at,
             "completed_at": completed_at,
             "inference_ms": inference_ms,

app/services/generation_provenance.py

@@ -14,6 +14,7 @@
 
 
 INITIATORS = frozenset({"user", "wizard", "system", "unknown"})
+_SUBMISSION_COMMAND_FIELDS = ("command_id", "workflow_id", "run_id")
 
 
 class CommandContext(TypedDict, total=False):
@@ -70,6 +71,37 @@ def resolve_generation_location(
     return {"workspace_id": None, "output_folder": None}
 
 
+def normalize_submission_provenance(value: Any) -> GenerationProvenance:
+    """Validate the optional provenance attached to a generation request.
+
+    This is attribution data, not an authorization boundary. Runtime-owned
+    identifiers (job/task/pipeline) and the physical output folder are added
+    by the backend and therefore cannot be supplied by the browser.
+    """
+    raw = value if isinstance(value, Mapping) else {}
+    actor = _clean(raw.get("actor")) or "unknown"
+    if actor not in INITIATORS:
+        actor = "unknown"
+    capability = _clean(raw.get("capability"))
+    workspace_id = _clean(raw.get("workspace_id"))
+    command_raw = raw.get("command") if isinstance(raw.get("command"), Mapping) else {}
+    command: CommandContext = {}
+    for key in _SUBMISSION_COMMAND_FIELDS:
+        cleaned = _clean(command_raw.get(key))
+        if cleaned:
+            command[key] = cleaned[:200]
+    result: GenerationProvenance = {
+        "actor": actor,
+        "tool": "studio",
+        "command": command,
+    }
+    if capability:
+        result["capability"] = capability[:200]
+    if workspace_id:
+        result["workspace_id"] = workspace_id[:200]
+    return result
+
+
 def provenance_from_manifest(manifest: Mapping[str, Any] | None) -> GenerationProvenance:
     """Project a canonical manifest onto initiator vs provider/model vs location."""
     value = manifest if isinstance(manifest, Mapping) else {}
@@ -101,5 +133,6 @@ def provenance_from_manifest(manifest: Mapping[str, Any] | None) -> GenerationPr
 
 __all__ = [
     "CommandContext", "GenerationLocation", "GenerationProvenance", "INITIATORS",
-    "provenance_from_manifest", "resolve_generation_location",
+    "normalize_submission_provenance", "provenance_from_manifest",
+    "resolve_generation_location",
 ]