From d3e6a5b1b9f6df3a6f1529cd89fcc82e5d33e79a Mon Sep 17 00:00:00 2001 From: Lightning Pixel Date: Fri, 13 Nov 2026 10:04:52 +0100 Subject: [PATCH 1/3] feat(runs): add a persisted run history store One RunRecord per generation, workflow or mesh-op run, kept in a JSON file at the workspace root. Writes are atomic, a corrupt file is moved aside, runs a previous backend left in flight are marked interrupted on load, and the history is capped. The record shape is the one provenance (T-F1) will reuse. --- api/schemas/runs.py | 62 ++++++ api/services/run_history.py | 364 ++++++++++++++++++++++++++++++++++++ 2 files changed, 426 insertions(+) create mode 100644 api/schemas/runs.py create mode 100644 api/services/run_history.py diff --git a/api/schemas/runs.py b/api/schemas/runs.py new file mode 100644 index 000000000..3d95902fc --- /dev/null +++ b/api/schemas/runs.py @@ -0,0 +1,62 @@ +from typing import Any, Literal, Optional +from pydantic import BaseModel, Field, computed_field + +RunKind = Literal["generation", "workflow", "mesh_op"] +# JobStatus' vocabulary plus "interrupted": a run that was still pending/running when +# the backend stopped. It is terminal -- nothing will ever finish it. +RunStatusValue = Literal["pending", "running", "done", "error", "cancelled", "interrupted"] + +TERMINAL_RUN_STATUSES = frozenset({"done", "error", "cancelled", "interrupted"}) + + +class RunStep(BaseModel): + """One timed phase of a run: a generation step label, or a workflow node.""" + name: str + node_id: Optional[str] = None + status: RunStatusValue = "running" + started_at: float # epoch seconds + ended_at: Optional[float] = None + + +class RunRecord(BaseModel): + """What ran, with what, for how long, and what came out of it. + + One record per run, whatever drove it: a backend generation job, a workflow + executed by the renderer, or a mesh operation. This is deliberately the shape + the provenance model (T-F1) is meant to reuse -- an output's provenance is the + record of the run that produced it (model or workflow, inputs, params, + output_path) -- so there is one store of "how was this made", not two. + + Timestamps are epoch seconds. ``output_path`` is workspace-relative (POSIX + separators) when the output lives in the workspace, absolute otherwise. + """ + id: str + kind: RunKind + status: RunStatusValue = "pending" + model_id: Optional[str] = None # generation model, e.g. "triposg/generate" + workflow_id: Optional[str] = None + workflow_name: Optional[str] = None + op_id: Optional[str] = None # mesh operation id, e.g. "decimate" + inputs: dict[str, Any] = Field(default_factory=dict) + params: dict[str, Any] = Field(default_factory=dict) + steps: list[RunStep] = Field(default_factory=list) + error: Optional[str] = None + output_path: Optional[str] = None + created_at: float + started_at: Optional[float] = None + ended_at: Optional[float] = None + + @computed_field # type: ignore[prop-decorator] + @property + def duration_s(self) -> Optional[float]: + start = self.started_at if self.started_at is not None else self.created_at + if self.ended_at is None: + return None + return round(max(0.0, self.ended_at - start), 3) + + +class RunRecordPage(BaseModel): + runs: list[RunRecord] + total: int + limit: int + offset: int diff --git a/api/services/run_history.py b/api/services/run_history.py new file mode 100644 index 000000000..77c7261b6 --- /dev/null +++ b/api/services/run_history.py @@ -0,0 +1,364 @@ +"""Persisted run history (one record per generation, workflow or mesh-op run). + +The live job table in ``routers.generation`` is in memory and purged 30 minutes +after a job ends, so on its own nothing survives a restart. This store keeps a +``RunRecord`` (see ``schemas.runs`` -- the shape the provenance model reuses) per +run in a single JSON file at the workspace root. + +Robustness rules: +- Writes are atomic (temp file + ``os.replace``) so a crash never leaves half a file. +- A missing file is an empty history; a corrupt one is moved aside and replaced. +- Records still pending/running in a file this process did not write them to were + left behind by a backend that stopped mid-run: they become ``interrupted``. +- The history is capped (``max_records``); the oldest finished runs go first. +- I/O failures are logged, never raised: history must not break a generation. + +The store is inert until ``start()`` gives it a path, so importing the routers +(tests, tooling) never writes into a real workspace. +""" +from __future__ import annotations + +import json +import os +import re +import tempfile +import threading +import time +from pathlib import Path +from typing import Any, Callable, Iterable, Optional + +from pydantic import ValidationError + +from schemas.runs import TERMINAL_RUN_STATUSES, RunKind, RunRecord, RunStep + +RUN_HISTORY_FILENAME = ".modly-run-history.json" +DEFAULT_MAX_RECORDS = 500 +MAX_STEPS = 200 +MAX_ERROR_CHARS = 4000 +_FILE_VERSION = 1 +INTERRUPTED_ERROR = "Interrupted: Modly stopped before this run finished" + +# Extensions report progress through the step label ("Sampling 12/50", "Denoising 40%"), +# so counters are folded out before deciding whether a label starts a new step. +_STEP_COUNTERS = re.compile(r"[\d.,%/]+") + + +def _step_key(name: str) -> str: + return _STEP_COUNTERS.sub("#", name).strip() + + +def _truncate_error(error: Optional[str]) -> Optional[str]: + if error is None or len(error) <= MAX_ERROR_CHARS: + return error + # Tracebacks end with the actual exception, so keep the tail. + return "…" + error[-(MAX_ERROR_CHARS - 1):] + + +class RunHistoryStore: + def __init__(self, max_records: int = DEFAULT_MAX_RECORDS) -> None: + self.max_records = max_records + self._lock = threading.RLock() + self._path_provider: Optional[Callable[[], Path]] = None + self._loaded_path: Optional[Path] = None + self._records: dict[str, RunRecord] = {} + # Runs this process is still driving; everything else non-terminal on disk is stale. + self._live: dict[str, RunRecord] = {} + + # ------------------------------------------------------------------ # + # Lifecycle + # ------------------------------------------------------------------ # + + def start(self, path_provider: Callable[[], Path]) -> None: + """Enable persistence. The provider is called per operation, so a relocated + workspace is followed. Loading marks runs a previous backend left behind as + interrupted.""" + with self._lock: + self._path_provider = path_provider + self._loaded_path = None + self._records = {} + self._live = {} + self._sync_path() + + def stop(self) -> None: + """Graceful shutdown: runs still in flight will never finish.""" + with self._lock: + if self._path_provider is None: + return + if self._live: + self._sync_path() + now = time.time() + for record in self._live.values(): + self._interrupt(record, now) + self._records[record.id] = record + self._live = {} + self._persist() + self._path_provider = None + self._loaded_path = None + self._records = {} + + @property + def enabled(self) -> bool: + return self._path_provider is not None + + # ------------------------------------------------------------------ # + # Recording + # ------------------------------------------------------------------ # + + def begin( + self, + run_id: str, + kind: RunKind, + *, + model_id: Optional[str] = None, + op_id: Optional[str] = None, + inputs: Optional[dict[str, Any]] = None, + params: Optional[dict[str, Any]] = None, + ) -> None: + """File a pending run (persisted, so a crash before it starts still shows).""" + if not self.enabled: + return + record = RunRecord( + id=run_id, kind=kind, status="pending", model_id=model_id, op_id=op_id, + inputs=inputs or {}, params=params or {}, + created_at=time.time(), + ) + self.save(record) + + def mark_running(self, run_id: str) -> None: + with self._lock: + record = self._live.get(run_id) + if record is None or record.status in TERMINAL_RUN_STATUSES: + return + record.status = "running" + record.started_at = record.started_at or time.time() + self.save(record) + + def step(self, run_id: str, name: str) -> None: + """Note the current step label. In memory only -- labels can change many times + a second; the steps are written out with the next status change.""" + if not name: + return + with self._lock: + record = self._live.get(run_id) + if record is None: + return + now = time.time() + current = record.steps[-1] if record.steps else None + if current is not None and current.ended_at is None: + if _step_key(current.name) == _step_key(name): + return + if len(record.steps) >= MAX_STEPS: + return # the last step absorbs the rest of the run + current.ended_at = now + current.status = "done" + record.steps.append(RunStep(name=name, started_at=now)) + + def finish( + self, + run_id: str, + status: str, + *, + error: Optional[str] = None, + output_path: Optional[str] = None, + ) -> None: + with self._lock: + record = self._live.get(run_id) + if record is None: + return + now = time.time() + record.status = status # type: ignore[assignment] + record.ended_at = now + if error is not None: + record.error = error + if output_path is not None: + record.output_path = output_path + _close_steps(record, now) + self.save(record) + + def save(self, record: RunRecord) -> None: + """Insert or replace a whole record (used directly for renderer-driven workflow + runs). A non-terminal record is tracked as live by this process.""" + if not self.enabled: + return + with self._lock: + self._sync_path() + record.inputs = _json_safe(record.inputs) + record.params = _json_safe(record.params) + record.error = _truncate_error(record.error) + if len(record.steps) > MAX_STEPS: + record.steps = record.steps[:MAX_STEPS] + if record.status in TERMINAL_RUN_STATUSES: + self._live.pop(record.id, None) + else: + self._live[record.id] = record + self._records[record.id] = record + self._apply_retention() + self._persist() + + # ------------------------------------------------------------------ # + # Reading + # ------------------------------------------------------------------ # + + def get(self, run_id: str) -> Optional[RunRecord]: + if not self.enabled: + return None + with self._lock: + self._sync_path() + record = self._records.get(run_id) + return record.model_copy(deep=True) if record is not None else None + + def list( + self, + *, + statuses: Optional[Iterable[str]] = None, + kinds: Optional[Iterable[str]] = None, + limit: int = 50, + offset: int = 0, + ) -> tuple[list[RunRecord], int]: + """Newest first. Returns (page, total matching).""" + if not self.enabled: + return [], 0 + wanted_status = set(statuses) if statuses else None + wanted_kind = set(kinds) if kinds else None + with self._lock: + self._sync_path() + matching = [ + r for r in self._records.values() + if (wanted_status is None or r.status in wanted_status) + and (wanted_kind is None or r.kind in wanted_kind) + ] + matching.sort(key=lambda r: r.created_at, reverse=True) + page = matching[offset:offset + limit] + return [r.model_copy(deep=True) for r in page], len(matching) + + # ------------------------------------------------------------------ # + # Internals (lock held) + # ------------------------------------------------------------------ # + + def _sync_path(self) -> None: + """(Re)load when the workspace path changed since the last load.""" + assert self._path_provider is not None + path = Path(self._path_provider()) + if path == self._loaded_path: + return + self._loaded_path = path + self._records = self._load(path) + now = time.time() + stale = False + for record in self._records.values(): + if record.status not in TERMINAL_RUN_STATUSES and record.id not in self._live: + self._interrupt(record, now) + stale = True + # Live runs keep being written wherever the workspace now is. + for run_id, record in self._live.items(): + self._records[run_id] = record + if stale: + self._persist() + + @staticmethod + def _interrupt(record: RunRecord, now: float) -> None: + record.status = "interrupted" + record.error = record.error or INTERRUPTED_ERROR + # When it really stopped is unknown; leave ended_at empty rather than invent one. + for step in record.steps: + if step.ended_at is None: + step.status = "interrupted" + + def _load(self, path: Path) -> dict[str, RunRecord]: + try: + raw = path.read_text(encoding="utf-8") + except FileNotFoundError: + return {} + except OSError as exc: + print(f"[RunHistory] Cannot read {path}: {exc}") + return {} + try: + data = json.loads(raw) + entries = data["runs"] if isinstance(data, dict) else None + if not isinstance(entries, list): + raise ValueError("missing 'runs' list") + except (ValueError, KeyError, TypeError) as exc: + self._quarantine(path, exc) + return {} + records: dict[str, RunRecord] = {} + for entry in entries: + try: + record = RunRecord.model_validate(entry) + except ValidationError: + continue # one bad record must not cost the whole history + records[record.id] = record + return records + + @staticmethod + def _quarantine(path: Path, exc: Exception) -> None: + target = path.with_name(f"{path.name}.corrupt-{int(time.time())}") + try: + os.replace(path, target) + print(f"[RunHistory] {path} is unreadable ({exc}); moved to {target.name}") + except OSError as move_exc: + print(f"[RunHistory] {path} is unreadable ({exc}) and could not be moved: {move_exc}") + + def _apply_retention(self) -> None: + excess = len(self._records) - self.max_records + if excess <= 0: + return + finished = sorted( + (r for r in self._records.values() if r.id not in self._live), + key=lambda r: r.created_at, + ) + for record in finished[:excess]: + del self._records[record.id] + + def _persist(self) -> None: + path = self._loaded_path + if path is None: + return + ordered = sorted(self._records.values(), key=lambda r: r.created_at, reverse=True) + payload = { + "version": _FILE_VERSION, + "runs": [r.model_dump(mode="json", exclude={"duration_s"}) for r in ordered], + } + tmp_name: Optional[str] = None + try: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_name = tempfile.mkstemp(prefix=f"{path.name}.", suffix=".tmp", dir=path.parent) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(payload, handle, ensure_ascii=False) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp_name, path) + tmp_name = None + except OSError as exc: + print(f"[RunHistory] Cannot write {path}: {exc}") + finally: + if tmp_name is not None: + try: + os.unlink(tmp_name) + except OSError: + pass + + +def _close_steps(record: RunRecord, now: float) -> None: + for step in record.steps: + if step.ended_at is None: + step.ended_at = now + step.status = record.status + + +def _json_safe(value: dict[str, Any]) -> dict[str, Any]: + """Params come from callers; keep only what survives a JSON round trip.""" + try: + return json.loads(json.dumps(value, default=str)) + except (TypeError, ValueError): + return {} + + +def workspace_relative(path: Path, workspace: Path) -> str: + """Workspace-relative POSIX path when inside the workspace, absolute otherwise.""" + try: + return path.resolve().relative_to(workspace.resolve()).as_posix() + except (OSError, ValueError): + return str(path) + + +run_history = RunHistoryStore() From adc686f52c71719a9c146b6947dc1db82deacd48 Mon Sep 17 00:00:00 2001 From: Lightning Pixel Date: Fri, 13 Nov 2026 10:04:53 +0100 Subject: [PATCH 2/3] feat(runs): record generations and mesh ops, list them at GET /workflow-runs Generation jobs are recorded when filed, started and ended (cancel included), with their step timings. Mesh ops that write a mesh or fail are recorded too. GET /workflow-runs pages the history with status/kind filters, PUT /workflow-runs/{id} lets the renderer report workflow runs, and GET /workflow-runs/{id} falls back to the history once a job is purged. --- api/main.py | 10 +- api/routers/generation.py | 37 ++- api/routers/optimize.py | 42 ++- api/routers/workflow_runs.py | 77 ++++- api/tests/test_generation_router.py | 2 + api/tests/test_run_history.py | 368 ++++++++++++++++++++++ api/tests/test_workflow_runs_lifecycle.py | 2 + api/tests/test_workflow_runs_router.py | 2 + 8 files changed, 532 insertions(+), 8 deletions(-) create mode 100644 api/tests/test_run_history.py diff --git a/api/main.py b/api/main.py index 958ba2a02..9bf352304 100644 --- a/api/main.py +++ b/api/main.py @@ -20,8 +20,16 @@ async def lifespan(app: FastAPI): # Startup: initialize the registry (instantiates all adapters) from services.generator_registry import generator_registry generator_registry.initialize() + # Persisted run history at the workspace root; read per call so a relocated + # workspace is followed. Loading marks runs a previous backend left running as + # interrupted. + import services.generator_registry as registry + from services.run_history import RUN_HISTORY_FILENAME, run_history + run_history.start(lambda: registry.WORKSPACE_DIR / RUN_HISTORY_FILENAME) yield - # Shutdown: unload all models and stop the local LLM server + # Shutdown: runs still in flight will never finish + run_history.stop() + # Unload all models and stop the local LLM server generator_registry.unload_all() from services.llm_server import llama_pool llama_pool.unload_all(force=True) diff --git a/api/routers/generation.py b/api/routers/generation.py index 33686cc3e..b1dd63b8c 100644 --- a/api/routers/generation.py +++ b/api/routers/generation.py @@ -17,6 +17,7 @@ import services.generator_registry as registry from services.generator_registry import generator_registry from schemas.generation import GenerateFromArtifactRequest, JobStatus +from services.run_history import run_history from services.artifact_input import ( RESERVED_ARTIFACT_PARAMS, TypedArtifactInput, @@ -56,6 +57,30 @@ def _purge_old_jobs() -> None: _completed_at.pop(jid, None) +def record_job_filed( + job_id: str, + model_id: str, + inputs: dict, + params: dict, + collection: str, +) -> None: + """File the persisted history record of a generation job (see services.run_history).""" + run_history.begin( + job_id, "generation", model_id=model_id, inputs=inputs, + params={**params, "collection": collection}, + ) + + +def _record_job_end(job_id: str) -> None: + """Copy a job's final state into its history record once its worker returned.""" + job = _jobs.get(job_id) + if job is None or job.status not in ("done", "error", "cancelled"): + run_history.finish(job_id, "error", error="Generation ended without a final status") + return + output_path = job.output_url.removeprefix("/workspace/") if job.output_url else None + run_history.finish(job_id, job.status, error=job.error, output_path=output_path) + + def sanitize_collection(collection: str) -> str: """Normalize a caller-supplied collection name into a safe workspace subfolder. @@ -146,6 +171,8 @@ async def generate_from_image( _jobs[job_id] = job _cancel_events[job_id] = threading.Event() + record_job_filed(job_id, model_id, {"image": image.filename}, full_params, collection) + background_tasks.add_task( _run_generation, job_id, image_bytes, full_params, collection, output_kind, model_id ) @@ -172,9 +199,12 @@ async def generate_from_artifact( 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()) + record_job_filed( + job_id, request.model_id, {request.input_kind: request.input_path}, params, collection + ) + params["scene_manifest_path"] = str(artifact.path) _purge_old_jobs() _jobs[job_id] = JobStatus(job_id=job_id, status="pending", progress=0) _cancel_events[job_id] = threading.Event() @@ -207,6 +237,7 @@ async def cancel_job(job_id: str): _cancel_events[job_id].set() job.status = "cancelled" _completed_at[job_id] = time.monotonic() + run_history.finish(job_id, "cancelled") # Kill only the subprocess bound to this job. A queued cancellation must not # terminate whichever earlier job currently owns the active generator, and an # unpinned job runs beside the pinned worker, so it can share its generator: @@ -274,6 +305,8 @@ async def _run_generation( job.status = "cancelled" _completed_at[job_id] = time.monotonic() raise + finally: + _record_job_end(job_id) def _run_generation_impl( @@ -289,6 +322,7 @@ def _run_generation_impl( job = _jobs[job_id] job.status = "running" job.step = None + run_history.mark_running(job_id) def progress_cb(pct: int, step: str = "") -> None: # Monotonic: the loading phase walks the bar up on a background thread and @@ -298,6 +332,7 @@ def progress_cb(pct: int, step: str = "") -> None: job.progress = pct if step: job.step = step + run_history.step(job_id, step) try: # Refuse a selected weight variant that is not installed before loading diff --git a/api/routers/optimize.py b/api/routers/optimize.py index 5fafc3343..c6d70a1c0 100644 --- a/api/routers/optimize.py +++ b/api/routers/optimize.py @@ -1,6 +1,7 @@ import hashlib import os import tempfile +import time import uuid from contextlib import contextmanager @@ -26,6 +27,8 @@ ) from services.mesh_ops.glb_patch import GlbPatchError, VertexPatch, patch_glb_file from services.mesh_ops.transforms import apply_transforms, transform_file +from schemas.runs import RunRecord +from services.run_history import run_history, workspace_relative router = APIRouter(tags=["optimize"]) @@ -131,8 +134,43 @@ def _run_operation( preserve_visuals: bool = False, ) -> MeshOpResult: context = _operation_context(output_path, preserve_visuals) - with _mesh_op_http_errors(): - return mesh_ops_registry.run(operation_id, input_path, params, context) + started_at = time.time() + try: + with _mesh_op_http_errors(): + result = mesh_ops_registry.run(operation_id, input_path, params, context) + except HTTPException as exc: + _record_mesh_op(operation_id, input_path, params, started_at, error=str(exc.detail)) + raise + # Analysis ops (inspect_mesh, suggest_fixes) only report: the viewer calls them on + # every selection, and they would crowd real runs out of the capped history. + if result.file_path is not None: + _record_mesh_op(operation_id, input_path, params, started_at, output=result.file_path) + return result + + +def _record_mesh_op( + operation_id: str, + input_path: Path, + params: dict[str, object], + started_at: float, + output: Path | None = None, + error: str | None = None, +) -> None: + """Persist a synchronous mesh op in run history (services.run_history).""" + workspace = registry.WORKSPACE_DIR + run_history.save(RunRecord( + id=str(uuid.uuid4()), + kind="mesh_op", + op_id=operation_id, + status="error" if error is not None else "done", + inputs={"mesh": workspace_relative(input_path, workspace)}, + params=dict(params), + error=error, + output_path=workspace_relative(output, workspace) if output is not None else None, + created_at=started_at, + started_at=started_at, + ended_at=time.time(), + )) def _operation_response(result: MeshOpResult) -> dict[str, object]: diff --git a/api/routers/workflow_runs.py b/api/routers/workflow_runs.py index 8a210ea2d..b928c9ae3 100644 --- a/api/routers/workflow_runs.py +++ b/api/routers/workflow_runs.py @@ -1,9 +1,10 @@ import json import threading +import time import uuid -from typing import Optional -from fastapi import APIRouter, BackgroundTasks, File, Form, HTTPException, UploadFile -from pydantic import BaseModel +from typing import Any, Optional +from fastapi import APIRouter, BackgroundTasks, File, Form, HTTPException, Query, UploadFile +from pydantic import BaseModel, Field from routers.generation import ( VALID_REMESH_MODES, @@ -12,10 +13,13 @@ _purge_old_jobs, _run_generation, cancel_job, + record_job_filed, sanitize_collection, ) from schemas.generation import JobStatus +from schemas.runs import RunRecord, RunRecordPage, RunStatusValue, RunStep, RunKind from services.generator_registry import generator_registry +from services.run_history import run_history router = APIRouter(tags=["workflow-runs"]) @@ -30,6 +34,55 @@ class WorkflowRunStatus(BaseModel): scene_candidate: Optional[dict] = None +class WorkflowRunRecordIn(BaseModel): + """A workflow run's record as the renderer, which executes workflows, reports it.""" + workflow_id: Optional[str] = None + workflow_name: Optional[str] = None + status: RunStatusValue + inputs: dict[str, Any] = Field(default_factory=dict) + params: dict[str, Any] = Field(default_factory=dict) + steps: list[RunStep] = Field(default_factory=list) + error: Optional[str] = None + output_path: Optional[str] = None + created_at: Optional[float] = None + started_at: Optional[float] = None + ended_at: Optional[float] = None + + +MAX_PAGE = 200 + + +@router.get("", response_model=RunRecordPage) +async def list_runs( + status: Optional[list[RunStatusValue]] = Query(None), + kind: Optional[list[RunKind]] = Query(None), + limit: int = Query(50, ge=1, le=MAX_PAGE), + offset: int = Query(0, ge=0), +): + """Persisted run history, newest first. ``status``/``kind`` may repeat to match any.""" + runs, total = run_history.list(statuses=status, kinds=kind, limit=limit, offset=offset) + return RunRecordPage(runs=runs, total=total, limit=limit, offset=offset) + + +@router.put("/{run_id}", response_model=RunRecord) +async def put_workflow_run_record(run_id: str, body: WorkflowRunRecordIn): + """Insert or replace the history record of a renderer-driven workflow run.""" + if not run_id or len(run_id) > 200: + raise HTTPException(400, "Invalid run id") + existing = run_history.get(run_id) + if existing is not None and existing.kind != "workflow": + raise HTTPException(409, f"Run {run_id} is not a workflow run") + record = RunRecord( + id=run_id, + kind="workflow", + created_at=body.created_at + or (existing.created_at if existing is not None else time.time()), + **body.model_dump(exclude={"created_at"}), + ) + run_history.save(record) + return record + + @router.post("/from-image") async def create_run_from_image( background_tasks: BackgroundTasks, @@ -93,6 +146,7 @@ async def create_run_from_image( _jobs[job_id] = JobStatus(job_id=job_id, status="pending", progress=0) _cancel_events[job_id] = threading.Event() + record_job_filed(job_id, model_id, {"image": image.filename}, full_params, collection) background_tasks.add_task( _run_generation, job_id, image_bytes, full_params, collection, output_kind, model_id @@ -105,7 +159,22 @@ async def create_run_from_image( async def get_run(run_id: str): job = _jobs.get(run_id) if not job: - raise HTTPException(404, f"Run {run_id} not found") + # Purged from the live table (or from before a restart): answer from history. + record = run_history.get(run_id) + if record is None: + raise HTTPException(404, f"Run {run_id} not found") + output_url = f"/workspace/{record.output_path}" if ( + record.output_path and record.status == "done" + and not record.output_path.startswith("/") and ":" not in record.output_path + ) else None + return WorkflowRunStatus( + run_id=record.id, + status=record.status, + progress=100 if record.status == "done" else 0, + output_url=output_url, + error=record.error, + scene_candidate={"workspace_path": record.output_path} if output_url else None, + ) scene_candidate = None if job.status == "done" and job.output_url: diff --git a/api/tests/test_generation_router.py b/api/tests/test_generation_router.py index 3d69118c7..784fc2248 100644 --- a/api/tests/test_generation_router.py +++ b/api/tests/test_generation_router.py @@ -26,6 +26,8 @@ def generate(self, image_bytes, params, progress_cb, cancel_event=None) -> Path: class _FakeUpload: """Minimal UploadFile stand-in: an image content-type and readable bytes.""" + filename = "input.png" + def __init__(self, content_type: str = "image/png", data: bytes = b"\x89PNG\r\n") -> None: self.content_type = content_type self._data = data diff --git a/api/tests/test_run_history.py b/api/tests/test_run_history.py new file mode 100644 index 000000000..8c67f95f4 --- /dev/null +++ b/api/tests/test_run_history.py @@ -0,0 +1,368 @@ +import asyncio +import json +import tempfile +import threading +import time +import unittest +from pathlib import Path +from unittest.mock import patch + +from fastapi import HTTPException + +import routers.generation as generation +import routers.optimize as optimize +import routers.workflow_runs as workflow_runs +import services.run_history as run_history_module +from schemas.generation import JobStatus +from schemas.runs import RunRecord, RunStep +from services.mesh_ops import MeshOpExecutionError, MeshOpResult +from services.run_history import ( + INTERRUPTED_ERROR, + MAX_STEPS, + RUN_HISTORY_FILENAME, + RunHistoryStore, +) + + +def _record(run_id: str, status: str = "done", created_at: float = 1000.0, kind: str = "generation") -> RunRecord: + return RunRecord(id=run_id, kind=kind, status=status, created_at=created_at) + + +class _TempStoreCase(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.workspace = Path(self._tmp.name) + self.path = self.workspace / RUN_HISTORY_FILENAME + self.store = RunHistoryStore() + self.store.start(lambda: self.path) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def reopen(self, max_records: int | None = None) -> RunHistoryStore: + """A fresh store on the same file: what the next app start sees.""" + store = RunHistoryStore(max_records or self.store.max_records) + store.start(lambda: self.path) + return store + + +class RunHistoryStoreTests(_TempStoreCase): + def test_inert_until_started(self) -> None: + store = RunHistoryStore() + store.begin("a", "generation") + store.save(_record("b")) + self.assertEqual(store.list(), ([], 0)) + self.assertIsNone(store.get("b")) + + def test_lifecycle_is_recorded_with_steps_and_duration(self) -> None: + self.store.begin("run-1", "generation", model_id="sf3d", inputs={"image": "a.png"}, params={"seed": 1}) + self.assertEqual(self.store.get("run-1").status, "pending") + self.store.mark_running("run-1") + self.store.step("run-1", "Loading SF3D…") + self.store.step("run-1", "Sampling 1/50") + self.store.step("run-1", "Sampling 2/50") # same step, only the counter moved + self.store.step("run-1", "Exporting") + self.store.finish("run-1", "done", output_path="Default/out.glb") + + record = self.reopen().get("run-1") + self.assertEqual(record.status, "done") + self.assertEqual(record.model_id, "sf3d") + self.assertEqual(record.inputs, {"image": "a.png"}) + self.assertEqual(record.params, {"seed": 1}) + self.assertEqual(record.output_path, "Default/out.glb") + self.assertEqual([s.name for s in record.steps], ["Loading SF3D…", "Sampling 1/50", "Exporting"]) + self.assertTrue(all(s.ended_at is not None and s.status == "done" for s in record.steps)) + self.assertIsNotNone(record.duration_s) + self.assertGreaterEqual(record.duration_s, 0) + + def test_steps_are_capped(self) -> None: + self.store.begin("r", "generation") + for i in range(MAX_STEPS + 20): + self.store.step("r", f"step {chr(65 + i % 26)}{chr(65 + i // 26)}") + self.store.finish("r", "done") + self.assertEqual(len(self.store.get("r").steps), MAX_STEPS) + + def test_error_and_cancel_are_final(self) -> None: + self.store.begin("bad", "generation") + self.store.finish("bad", "error", error="x" * 10_000) + self.store.begin("stop", "generation") + self.store.finish("stop", "cancelled") + self.store.finish("stop", "done") # a late worker return must not overwrite it + + reopened = self.reopen() + bad = reopened.get("bad") + self.assertEqual(bad.status, "error") + self.assertLessEqual(len(bad.error), run_history_module.MAX_ERROR_CHARS) + self.assertEqual(reopened.get("stop").status, "cancelled") + + def test_list_is_newest_first_with_paging(self) -> None: + for i in range(5): + self.store.save(_record(f"r{i}", created_at=1000.0 + i)) + page, total = self.store.list(limit=2, offset=1) + self.assertEqual(total, 5) + self.assertEqual([r.id for r in page], ["r3", "r2"]) + page, total = self.store.list(limit=10, offset=4) + self.assertEqual([r.id for r in page], ["r0"]) + + def test_list_filters_by_status_and_kind(self) -> None: + self.store.save(_record("ok", "done", 1)) + self.store.save(_record("bad", "error", 2)) + self.store.save(_record("stop", "cancelled", 3, kind="workflow")) + page, total = self.store.list(statuses=["error", "cancelled"]) + self.assertEqual((total, [r.id for r in page]), (2, ["stop", "bad"])) + page, total = self.store.list(kinds=["workflow"]) + self.assertEqual([r.id for r in page], ["stop"]) + + def test_retention_cap_drops_oldest_finished_runs(self) -> None: + store = RunHistoryStore(max_records=3) + store.start(lambda: self.path) + store.begin("live", "generation") # oldest, but still running: kept + store.mark_running("live") + for i in range(4): + store.save(_record(f"r{i}", created_at=time.time() + 1 + i)) + ids = {r.id for r in store.list(limit=50)[0]} + self.assertEqual(ids, {"live", "r2", "r3"}) + on_disk = json.loads(self.path.read_text(encoding="utf-8"))["runs"] + self.assertEqual(len(on_disk), 3) + + def test_restart_marks_in_flight_runs_interrupted(self) -> None: + self.store.begin("queued", "generation") + self.store.begin("busy", "generation") + self.store.mark_running("busy") + self.store.step("busy", "Generating") + self.store.save(_record("finished", "done")) + + reopened = self.reopen() # the old process died without stop() + for run_id in ("queued", "busy"): + record = reopened.get(run_id) + self.assertEqual(record.status, "interrupted") + self.assertEqual(record.error, INTERRUPTED_ERROR) + self.assertEqual(reopened.get("finished").status, "done") + # ... and it is written back, so the next reader agrees. + raw = json.loads(self.path.read_text(encoding="utf-8")) + self.assertEqual({r["id"]: r["status"] for r in raw["runs"]}["busy"], "interrupted") + + def test_stop_marks_live_runs_interrupted(self) -> None: + self.store.begin("busy", "generation") + self.store.mark_running("busy") + self.store.stop() + self.assertEqual(self.reopen().get("busy").status, "interrupted") + + def test_missing_file_is_an_empty_history(self) -> None: + self.assertEqual(self.store.list(), ([], 0)) + self.assertFalse(self.path.exists()) + + def test_corrupt_file_is_moved_aside(self) -> None: + self.path.write_text("{not json", encoding="utf-8") + store = self.reopen() + self.assertEqual(store.list(), ([], 0)) + store.save(_record("fresh")) + self.assertEqual(self.reopen().get("fresh").status, "done") + self.assertEqual(len(list(self.workspace.glob(f"{RUN_HISTORY_FILENAME}.corrupt-*"))), 1) + + def test_invalid_entries_are_skipped(self) -> None: + good = _record("good").model_dump(mode="json") + self.path.write_text(json.dumps({"version": 1, "runs": [good, {"id": "x"}, 42]}), encoding="utf-8") + page, total = self.reopen().list() + self.assertEqual([r.id for r in page], ["good"]) + + def test_writes_are_atomic_and_leave_no_temp_files(self) -> None: + for i in range(3): + self.store.save(_record(f"r{i}")) + self.assertEqual(sorted(p.name for p in self.workspace.iterdir()), [RUN_HISTORY_FILENAME]) + + def test_write_failure_does_not_raise(self) -> None: + with patch("services.run_history.os.replace", side_effect=PermissionError("locked")): + self.store.save(_record("r")) # logged, not raised + self.assertEqual(self.store.get("r").status, "done") # still served from memory + + def test_follows_a_relocated_workspace(self) -> None: + other = self.workspace / "moved" / RUN_HISTORY_FILENAME + current = {"path": self.path} + store = RunHistoryStore() + store.start(lambda: current["path"]) + store.begin("live", "generation") + store.mark_running("live") + current["path"] = other + store.finish("live", "done") + reloaded = RunHistoryStore() + reloaded.start(lambda: other) + self.assertEqual(reloaded.get("live").status, "done") + + +class _RouterHistoryCase(_TempStoreCase): + """Points the routers' shared store at a temp workspace for one test.""" + + def setUp(self) -> None: + super().setUp() + patcher = patch.object(run_history_module.run_history, "_path_provider", None) + patcher.start() + self.addCleanup(patcher.stop) + self.shared = run_history_module.run_history + self.shared.start(lambda: self.path) + self.addCleanup(self.shared.stop) + for store in (generation._jobs, generation._cancel_events, generation._cancelled, + generation._completed_at, generation._job_generators): + store.clear() + + +class ListRunsEndpointTests(_RouterHistoryCase): + def test_lists_with_paging_and_status_filter(self) -> None: + for i, status in enumerate(["done", "error", "cancelled", "interrupted", "done"]): + self.shared.save(_record(f"r{i}", status, 1000.0 + i)) + page = asyncio.run(workflow_runs.list_runs(status=["error", "cancelled"], kind=None, limit=1, offset=0)) + self.assertEqual(page.total, 2) + self.assertEqual([r.id for r in page.runs], ["r2"]) + page = asyncio.run(workflow_runs.list_runs(status=None, kind=None, limit=50, offset=0)) + self.assertEqual(page.total, 5) + self.assertEqual(page.runs[0].id, "r4") + + def test_route_is_mounted_at_the_collection_root(self) -> None: + paths = {(route.path, tuple(sorted(route.methods))) for route in workflow_runs.router.routes} + self.assertIn(("", ("GET",)), paths) + self.assertIn(("/{run_id}", ("PUT",)), paths) + + def test_get_run_falls_back_to_history_after_purge(self) -> None: + record = _record("old", "done") + record.output_path = "Default/out.glb" + self.shared.save(record) + status = asyncio.run(workflow_runs.get_run("old")) + self.assertEqual(status.status, "done") + self.assertEqual(status.output_url, "/workspace/Default/out.glb") + with self.assertRaises(HTTPException): + asyncio.run(workflow_runs.get_run("unknown")) + + +class PutWorkflowRunTests(_RouterHistoryCase): + def _put(self, run_id: str, **fields) -> RunRecord: + body = workflow_runs.WorkflowRunRecordIn(**fields) + return asyncio.run(workflow_runs.put_workflow_run_record(run_id, body)) + + def test_upserts_a_workflow_record(self) -> None: + self._put("wf-run", workflow_id="wf", workflow_name="Flow", status="running", + created_at=1000.0, started_at=1000.0, + steps=[RunStep(name="Mesh", node_id="n1", started_at=1000.0)]) + self._put("wf-run", workflow_id="wf", workflow_name="Flow", status="done", + started_at=1000.0, ended_at=1012.5, output_path="Workflows/out.glb", + steps=[RunStep(name="Mesh", node_id="n1", status="done", started_at=1000.0, ended_at=1012.5)]) + record = self.reopen().get("wf-run") + self.assertEqual(record.kind, "workflow") + self.assertEqual(record.status, "done") + self.assertEqual(record.created_at, 1000.0) # kept from the first report + self.assertEqual(record.duration_s, 12.5) + self.assertEqual(record.output_path, "Workflows/out.glb") + + def test_running_workflow_is_interrupted_on_restart(self) -> None: + self._put("wf-run", status="running", created_at=1000.0) + self.assertEqual(self.reopen().get("wf-run").status, "interrupted") + + def test_refuses_to_overwrite_a_backend_run(self) -> None: + self.shared.save(_record("gen")) + with self.assertRaises(HTTPException) as ctx: + self._put("gen", status="done") + self.assertEqual(ctx.exception.status_code, 409) + + +class _Gen: + def __init__(self, fail: bool = False, out_dir: Path | None = None) -> None: + self.fail = fail + self.outputs_dir = out_dir + + def generate(self, image_bytes, params, progress_cb, cancel_event=None): + progress_cb(10, "Generating") + if self.fail: + raise RuntimeError("boom") + out = Path(self.outputs_dir) / "mesh.glb" + out.write_bytes(b"glb") + return str(out) + + +class _Registry: + def __init__(self, gen: _Gen) -> None: + self.gen = gen + + def assert_weight_variant_installed(self, params, model_id) -> None: + pass + + def get_generator(self, model_id): + return self.gen + + def activate_ready_generator(self, model_id): + return self.gen + + def model_status(self, model_id): + return {"loaded": True, "name": model_id, "downloaded": True} + + +class GenerationRecordingTests(_RouterHistoryCase): + def setUp(self) -> None: + super().setUp() + ws = patch.object(generation.registry, "WORKSPACE_DIR", self.workspace) + ws.start() + self.addCleanup(ws.stop) + + def _run(self, job_id: str, gen: _Gen) -> None: + generation._jobs[job_id] = JobStatus(job_id=job_id, status="pending") + generation._cancel_events[job_id] = threading.Event() + generation.record_job_filed(job_id, "demo/model", {"image": "in.png"}, {"seed": 3}, "Default") + with patch.object(generation, "generator_registry", _Registry(gen)): + asyncio.run(generation._run_generation(job_id, b"img", {"seed": 3}, "Default", "mesh", "demo/model")) + + def test_successful_generation_is_recorded_done(self) -> None: + self._run("ok", _Gen()) + record = self.reopen().get("ok") + self.assertEqual(record.status, "done") + self.assertEqual(record.model_id, "demo/model") + self.assertEqual(record.params, {"seed": 3, "collection": "Default"}) + self.assertEqual(record.output_path, "Default/mesh.glb") + self.assertEqual([s.name for s in record.steps], ["Generating"]) + self.assertIsNotNone(record.duration_s) + + def test_failed_generation_is_recorded_with_its_error(self) -> None: + self._run("bad", _Gen(fail=True)) + record = self.reopen().get("bad") + self.assertEqual(record.status, "error") + self.assertIn("boom", record.error) + + def test_cancel_is_recorded_immediately(self) -> None: + generation._jobs["c"] = JobStatus(job_id="c", status="running") + generation.record_job_filed("c", "demo/model", {}, {}, "Default") + asyncio.run(generation.cancel_job("c")) + self.assertEqual(self.reopen().get("c").status, "cancelled") + + +class MeshOpRecordingTests(_RouterHistoryCase): + def setUp(self) -> None: + super().setUp() + ws = patch.object(optimize.registry, "WORKSPACE_DIR", self.workspace) + ws.start() + self.addCleanup(ws.stop) + self.mesh = self.workspace / "Default" / "in.glb" + self.mesh.parent.mkdir() + self.mesh.write_bytes(b"glb") + + def test_mesh_writing_op_is_recorded(self) -> None: + out = self.workspace / "Default" / "out.glb" + with patch.object(optimize.mesh_ops_registry, "run", return_value=MeshOpResult(file_path=out)): + optimize._run_operation("decimate", self.mesh, {"target_faces": 100}, out) + page, _ = self.reopen().list() + self.assertEqual(len(page), 1) + self.assertEqual((page[0].kind, page[0].op_id, page[0].status), ("mesh_op", "decimate", "done")) + self.assertEqual(page[0].inputs, {"mesh": "Default/in.glb"}) + self.assertEqual(page[0].output_path, "Default/out.glb") + + def test_failed_op_is_recorded(self) -> None: + with patch.object(optimize.mesh_ops_registry, "run", side_effect=MeshOpExecutionError("broken")): + with self.assertRaises(HTTPException): + optimize._run_operation("smooth", self.mesh, {}) + page, _ = self.reopen().list() + self.assertEqual((page[0].status, page[0].error), ("error", "broken")) + + def test_report_only_op_is_not_recorded(self) -> None: + with patch.object(optimize.mesh_ops_registry, "run", return_value=MeshOpResult(file_path=None)): + optimize._run_operation("inspect_mesh", self.mesh, {}) + self.assertEqual(self.reopen().list(), ([], 0)) + + +if __name__ == "__main__": + unittest.main() diff --git a/api/tests/test_workflow_runs_lifecycle.py b/api/tests/test_workflow_runs_lifecycle.py index cf1f5269b..f80a7590f 100644 --- a/api/tests/test_workflow_runs_lifecycle.py +++ b/api/tests/test_workflow_runs_lifecycle.py @@ -18,6 +18,8 @@ class _FakeUpload: """Minimal UploadFile stand-in: an image content-type and readable bytes.""" + filename = "input.png" + def __init__(self, content_type: str = "image/png", data: bytes = b"\x89PNG\r\n") -> None: self.content_type = content_type self._data = data diff --git a/api/tests/test_workflow_runs_router.py b/api/tests/test_workflow_runs_router.py index cb2bb210f..3515ff532 100644 --- a/api/tests/test_workflow_runs_router.py +++ b/api/tests/test_workflow_runs_router.py @@ -10,6 +10,8 @@ class _FakeUpload: """Minimal stand-in for UploadFile: an image content-type and readable bytes.""" + filename = "input.png" + def __init__(self, content_type: str = "image/png", data: bytes = b"\x89PNG\r\n") -> None: self.content_type = content_type self._data = data From 2afa5407f7319bd828d81bbb913eb4909a42d47b Mon Sep 17 00:00:00 2001 From: Lightning Pixel Date: Fri, 13 Nov 2026 10:04:55 +0100 Subject: [PATCH 3/3] feat(runs): persist workflow runs from the renderer The workflow runner reports its run record (per-node step timings, final status, output) to PUT /workflow-runs/{id}, chaining writes so the final status always lands last. Cancelled and abandoned runs are recorded as cancelled. --- package.json | 2 +- src/areas/workflows/workflowRunRecord.test.ts | 95 +++++++++++++++++++ src/areas/workflows/workflowRunRecord.ts | 95 +++++++++++++++++++ src/areas/workflows/workflowRunStore.ts | 66 +++++++++++++ src/shared/types/runs.ts | 44 +++++++++ 5 files changed, 301 insertions(+), 1 deletion(-) create mode 100644 src/areas/workflows/workflowRunRecord.test.ts create mode 100644 src/areas/workflows/workflowRunRecord.ts create mode 100644 src/shared/types/runs.ts diff --git a/package.json b/package.json index 74c026e63..f7c15b511 100644 --- a/package.json +++ b/package.json @@ -13,7 +13,7 @@ "prepare-resources": "node scripts/download-python-embed.js", "test": "npm run test:py && npm run test:node", "test:py": "node scripts/run-pytests.mjs", - "test:node": "node --test --experimental-strip-types --experimental-loader ./scripts/node-ts-extensionless-loader.mjs src/shared/types/assetLibrary.test.ts src/areas/generate/assetLibraryProjection.test.ts src/areas/generate/assetLibraryService.test.ts src/areas/generate/assetLibraryUi.test.ts src/areas/library/libraryView.test.ts src/areas/library/gridWindow.test.ts src/areas/generate/orcaSlicerLink.test.ts src/shared/utils/agentStream.test.ts src/shared/thumbnails/assetThumbnailQueue.test.ts src/shared/thumbnails/assetThumbnailStore.test.ts src/shared/components/agent/clientTools.test.ts src/shared/components/agent/agentContext.test.ts src/shared/utils/ollamaHealth.test.ts src/shared/utils/notification.test.ts src/shared/utils/agentConversations.test.ts electron/main/artifact-registry-service.test.ts electron/main/asset-thumbnail-cache.test.ts electron/main/extension-path-guard.test.ts electron/main/storage-directory-guard.test.ts electron/main/storage-ownership.test.ts electron/main/startup-error.test.ts electron/main/api-port.test.ts electron/main/agent-conversations.test.ts electron/preload/artifact-registry-preload.test.ts src/shared/stores/meshHistory.test.ts src/areas/generate/meshEditPanel.test.ts src/areas/generate/sculpt/SculptMesh.test.ts src/areas/generate/sculpt/SculptMesh.perf.test.ts src/areas/generate/sculpt/sculptLogic.test.ts src/shared/pages/pageRegistry.test.ts && node --test electron/main/*.test.mjs src/**/*.test.mjs", + "test:node": "node --test --experimental-strip-types --experimental-loader ./scripts/node-ts-extensionless-loader.mjs src/shared/types/assetLibrary.test.ts src/areas/generate/assetLibraryProjection.test.ts src/areas/generate/assetLibraryService.test.ts src/areas/generate/assetLibraryUi.test.ts src/areas/library/libraryView.test.ts src/areas/library/gridWindow.test.ts src/areas/generate/orcaSlicerLink.test.ts src/shared/utils/agentStream.test.ts src/shared/thumbnails/assetThumbnailQueue.test.ts src/shared/thumbnails/assetThumbnailStore.test.ts src/shared/components/agent/clientTools.test.ts src/shared/components/agent/agentContext.test.ts src/shared/utils/ollamaHealth.test.ts src/shared/utils/notification.test.ts src/shared/utils/agentConversations.test.ts electron/main/artifact-registry-service.test.ts electron/main/asset-thumbnail-cache.test.ts electron/main/extension-path-guard.test.ts electron/main/storage-directory-guard.test.ts electron/main/storage-ownership.test.ts electron/main/startup-error.test.ts electron/main/api-port.test.ts electron/main/agent-conversations.test.ts electron/preload/artifact-registry-preload.test.ts src/shared/stores/meshHistory.test.ts src/areas/generate/meshEditPanel.test.ts src/areas/generate/sculpt/SculptMesh.test.ts src/areas/generate/sculpt/SculptMesh.perf.test.ts src/areas/generate/sculpt/sculptLogic.test.ts src/shared/pages/pageRegistry.test.ts src/areas/workflows/workflowRunRecord.test.ts && node --test electron/main/*.test.mjs src/**/*.test.mjs", "package": "cross-env CSC_IDENTITY_AUTO_DISCOVERY=false npm run build && npm run prepare-resources && electron-builder", "package:mac": "cross-env CSC_IDENTITY_AUTO_DISCOVERY=false npm run build && npm run prepare-resources && electron-builder --mac --arm64", "lint": "eslint ." diff --git a/src/areas/workflows/workflowRunRecord.test.ts b/src/areas/workflows/workflowRunRecord.test.ts new file mode 100644 index 000000000..099a608e9 --- /dev/null +++ b/src/areas/workflows/workflowRunRecord.test.ts @@ -0,0 +1,95 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import { + MAX_RECORD_STEPS, + endRecordStep, + finishWorkflowRunRecord, + newWorkflowRunRecord, + reopenWorkflowRunRecord, + runOutputPath, + startRecordStep, + workflowRunParams, +} from './workflowRunRecord' + +const workflow = { + id: 'wf-1', + name: 'Image to mesh', + nodes: [ + { id: 'img', type: 'imageNode', data: { enabled: true, params: { filePath: 'C:/a.png' } } }, + { id: 'gen', type: 'extensionNode', data: { extensionId: 'triposg/generate', enabled: true, params: { seed: 7 } } }, + { id: 'off', type: 'extensionNode', data: { extensionId: 'mesh/decimate', enabled: false, params: {} } }, + ], +} + +test('a new record is running with the workflow, its inputs and enabled node params', () => { + const record = newWorkflowRunRecord('run-1', workflow, { image: 'C:/a.png' }, 100) + assert.equal(record.status, 'running') + assert.equal(record.workflow_id, 'wf-1') + assert.equal(record.workflow_name, 'Image to mesh') + assert.deepEqual(record.inputs, { image: 'C:/a.png' }) + assert.deepEqual(record.params, { nodes: { gen: { extension_id: 'triposg/generate', params: { seed: 7 } } } }) + assert.equal(record.created_at, 100) + assert.equal(record.started_at, 100) + assert.deepEqual(record.steps, []) +}) + +test('workflowRunParams skips disabled and non-extension nodes', () => { + assert.deepEqual(Object.keys(workflowRunParams(workflow).nodes as object), ['gen']) +}) + +test('steps are timed per node and a new step closes the open one as done', () => { + let record = newWorkflowRunRecord('run-1', workflow, {}, 100) + record = startRecordStep(record, 'TripoSG', 'gen', 101) + record = startRecordStep(record, 'Decimate', 'dec', 110) + assert.deepEqual(record.steps[0], { name: 'TripoSG', node_id: 'gen', status: 'done', started_at: 101, ended_at: 110 }) + assert.deepEqual(record.steps[1], { name: 'Decimate', node_id: 'dec', status: 'running', started_at: 110 }) + record = endRecordStep(record, 'error', 115) + assert.equal(record.steps[1].status, 'error') + assert.equal(record.steps[1].ended_at, 115) + // Ending again is a no-op: the step is already closed. + assert.equal(endRecordStep(record, 'done', 120), record) +}) + +test('steps are capped', () => { + let record = newWorkflowRunRecord('run-1', workflow, {}, 0) + for (let i = 0; i < MAX_RECORD_STEPS + 10; i++) record = startRecordStep(record, `n${i}`, `n${i}`, i) + assert.equal(record.steps.length, MAX_RECORD_STEPS) +}) + +test('finishing closes the open step with the run status and records the outcome', () => { + let record = startRecordStep(newWorkflowRunRecord('run-1', workflow, {}, 100), 'TripoSG', 'gen', 100) + record = finishWorkflowRunRecord(record, 'error', 130, { error: 'Error: boom' }) + assert.equal(record.status, 'error') + assert.equal(record.ended_at, 130) + assert.equal(record.error, 'Error: boom') + assert.equal(record.steps[0].status, 'error') + + const done = finishWorkflowRunRecord(newWorkflowRunRecord('run-2', workflow, {}, 100), 'done', 150, { outputPath: 'Workflows/a.glb' }) + assert.equal(done.output_path, 'Workflows/a.glb') + assert.equal(done.error, null) +}) + +test('a cancelled run keeps its status and partial steps', () => { + let record = startRecordStep(newWorkflowRunRecord('run-1', workflow, {}, 100), 'TripoSG', 'gen', 100) + record = finishWorkflowRunRecord(record, 'cancelled', 105) + assert.equal(record.status, 'cancelled') + assert.equal(record.steps[0].status, 'cancelled') +}) + +test('reopening a finished run makes it running again and clears the end', () => { + const finished = finishWorkflowRunRecord(newWorkflowRunRecord('run-1', workflow, {}, 100), 'error', 120, { error: 'x' }) + const reopened = reopenWorkflowRunRecord(finished) + assert.equal(reopened.status, 'running') + assert.equal(reopened.ended_at, null) + assert.equal(reopened.error, null) + const running = newWorkflowRunRecord('run-2', workflow, {}, 100) + assert.equal(reopenWorkflowRunRecord(running), running) +}) + +test('runOutputPath prefers a workspace-relative path', () => { + assert.equal(runOutputPath('/workspace/Workflows/a 100%.glb'), 'Workflows/a 100%.glb') + assert.equal(runOutputPath(undefined, 'D:/out/a.wav'), 'D:/out/a.wav') + assert.equal(runOutputPath('/optimize/serve-file?path=x', 'D:/x.glb'), 'D:/x.glb') + assert.equal(runOutputPath(), undefined) +}) diff --git a/src/areas/workflows/workflowRunRecord.ts b/src/areas/workflows/workflowRunRecord.ts new file mode 100644 index 000000000..e236e823e --- /dev/null +++ b/src/areas/workflows/workflowRunRecord.ts @@ -0,0 +1,95 @@ +// Pure bookkeeping for a workflow run's history record (see shared/types/runs.ts). +// The renderer executes workflows, so it owns their record and reports it with +// PUT /workflow-runs/{id}; the store wires these helpers to the run lifecycle. +// Every helper returns a new record. `now` is epoch seconds. + +import type { RunRecord, RunStatus } from '../../shared/types/runs' + +export type WorkflowRunRecord = Omit + +/** Matches the backend cap (services/run_history.py MAX_STEPS). */ +export const MAX_RECORD_STEPS = 200 + +const TERMINAL: ReadonlySet = new Set(['done', 'error', 'cancelled', 'interrupted']) + +export function isTerminalRunStatus(status: RunStatus): boolean { + return TERMINAL.has(status) +} + +interface WorkflowLike { + id: string + name: string + nodes: { id: string; type: string; data: { extensionId?: string; enabled?: boolean; params?: Record } }[] +} + +/** The params a run used: each enabled extension node's id, extension and params. */ +export function workflowRunParams(workflow: WorkflowLike): Record { + const nodes: Record = {} + for (const node of workflow.nodes) { + if (node.type !== 'extensionNode' || !node.data.enabled) continue + nodes[node.id] = { extension_id: node.data.extensionId ?? null, params: node.data.params ?? {} } + } + return { nodes } +} + +export function newWorkflowRunRecord( + id: string, + workflow: WorkflowLike, + inputs: Record, + now: number, +): WorkflowRunRecord { + return { + id, + status: 'running', + workflow_id: workflow.id, + workflow_name: workflow.name, + inputs, + params: workflowRunParams(workflow), + steps: [], + created_at: now, + started_at: now, + } +} + +/** Close the open step (if any) with `status`. */ +export function endRecordStep(record: WorkflowRunRecord, status: RunStatus, now: number): WorkflowRunRecord { + const last = record.steps[record.steps.length - 1] + if (!last || last.ended_at != null) return record + return { ...record, steps: [...record.steps.slice(0, -1), { ...last, status, ended_at: now }] } +} + +/** Open a step for a node; a step left open is closed as done first. */ +export function startRecordStep(record: WorkflowRunRecord, name: string, nodeId: string, now: number): WorkflowRunRecord { + const closed = endRecordStep(record, 'done', now) + if (closed.steps.length >= MAX_RECORD_STEPS) return closed + return { ...closed, steps: [...closed.steps, { name, node_id: nodeId, status: 'running', started_at: now }] } +} + +export function finishWorkflowRunRecord( + record: WorkflowRunRecord, + status: RunStatus, + now: number, + outcome: { error?: string; outputPath?: string } = {}, +): WorkflowRunRecord { + const closed = endRecordStep(record, status, now) + return { + ...closed, + status, + ended_at: now, + error: outcome.error ?? closed.error ?? null, + output_path: outcome.outputPath ?? closed.output_path ?? null, + } +} + +/** A finished run the user resumes (re-running a Wait branch) is running again. */ +export function reopenWorkflowRunRecord(record: WorkflowRunRecord): WorkflowRunRecord { + if (!isTerminalRunStatus(record.status)) return record + return { ...record, status: 'running', ended_at: null, error: null } +} + +/** Workspace-relative path of a `/workspace/...` URL, else the absolute path as-is. */ +export function runOutputPath(outputUrl?: string, outputPath?: string): string | undefined { + // toWorkspaceUrl does not percent-encode, so the rest of the URL is the path itself. + if (outputUrl?.startsWith('/workspace/')) return outputUrl.slice('/workspace/'.length) + return outputPath +} diff --git a/src/areas/workflows/workflowRunStore.ts b/src/areas/workflows/workflowRunStore.ts index 383ce0c69..87076f629 100644 --- a/src/areas/workflows/workflowRunStore.ts +++ b/src/areas/workflows/workflowRunStore.ts @@ -9,6 +9,12 @@ import type { Workflow, WFNode, WFEdge } from '@shared/types/electron.d' import { isBranchStarter, isSceneOutput, resolveDataSource, reachesSceneOutput, nearestUpstreamWaits } from './nodeBehaviors' import { assignSlotFilePaths } from './slotInputs' import type { SlotInputType } from './slotInputs' +import { + endRecordStep, finishWorkflowRunRecord, isTerminalRunStatus, newWorkflowRunRecord, + reopenWorkflowRunRecord, runOutputPath, startRecordStep, +} from './workflowRunRecord' +import type { WorkflowRunRecord } from './workflowRunRecord' +import type { RunStatus } from '@shared/types/runs' // ─── Types ──────────────────────────────────────────────────────────────────── @@ -42,6 +48,44 @@ const _pauseRequested = { current: false } // latest values when a body node starts (instead of the snapshot from run start). const _liveParams = { current: new Map>() } +// ─── Run history ────────────────────────────────────────────────────────────── +// The run's persisted record (GET /workflow-runs lists it). Every change is PUT to +// the backend; the PUTs are chained so a slow early write can never land after — +// and overwrite — the final status. + +const _record = { current: null as WorkflowRunRecord | null } +let _recordWrites: Promise = Promise.resolve() + +function nowSeconds(): number { + return Date.now() / 1000 +} + +function updateRecord(fn: (record: WorkflowRunRecord) => WorkflowRunRecord): void { + if (!_record.current) return + const record = fn(_record.current) + _record.current = record + const { id, ...body } = record + const apiUrl = useAppStore.getState().apiUrl + _recordWrites = _recordWrites + .then(() => axios.put(`${apiUrl}/workflow-runs/${encodeURIComponent(id)}`, body)) + .then(() => undefined, (err) => { console.warn('[workflows] Could not save run history:', err) }) +} + +function finishRecord(status: RunStatus, outcome?: { error?: string; outputPath?: string }): void { + if (!_record.current || isTerminalRunStatus(_record.current.status)) return + updateRecord((r) => finishWorkflowRunRecord(r, status, nowSeconds(), outcome)) +} + +function recordStepStart(node: WFNode, allExtensions: WorkflowExtension[]): void { + const name = getWorkflowExtension(node.data.extensionId ?? '', allExtensions)?.name + ?? node.data.extensionId ?? node.type + updateRecord((r) => startRecordStep(r, name, node.id, nowSeconds())) +} + +function recordStepEnd(status: RunStatus): void { + updateRecord((r) => endRecordStep(r, status, nowSeconds())) +} + function flushResume(): void { const fn = _resume.current _resume.current = null @@ -644,6 +688,7 @@ export const useWorkflowRunStore = create((set, get) => { }, })) useAppStore.getState().updateCurrentJob({ status: 'done', progress: 100, outputUrl }) + finishRecord('done', { outputPath: runOutputPath(outputUrl, outputPath) }) void showCompletionNotification('Workflow run complete') } @@ -666,6 +711,16 @@ export const useWorkflowRunStore = create((set, get) => { const appState = useAppStore.getState() const apiUrl = appState.apiUrl + // A previous run still parked on its Wait branches is abandoned by this one. + finishRecord('cancelled') + _record.current = newWorkflowRunRecord( + crypto.randomUUID(), + workflow, + overrideImageData ? { image: '(attached image)' } : appState.selectedImagePath ? { image: appState.selectedImagePath } : {}, + nowSeconds(), + ) + updateRecord((r) => r) + const { preExecExtNodes, branches, waitIds, parentWait, ordered } = identifyBranches(workflow) const branchSteps = waitIds.reduce((acc, w) => acc + (branches.get(w)?.length ?? 0), 0) @@ -680,6 +735,7 @@ export const useWorkflowRunStore = create((set, get) => { const dir = (w.data.params?.dir as string | undefined)?.trim() const fail = (msg: string, step: string): void => { set((s) => ({ runState: { ...s.runState, status: 'error', error: msg, blockStep: step }, activeNodeId: null })) + finishRecord('error', { error: msg }) } if (!dir) { fail('For Each: pick a folder first', 'No folder selected'); return } try { @@ -951,7 +1007,9 @@ export const useWorkflowRunStore = create((set, get) => { activeNodeId: node.id, runState: { ...s.runState, blockIndex: stepsDone, blockProgress: 0, blockStep: 'Starting…' }, })) + recordStepStart(node, allExtensions) await executeExtensionNode(node, ctx, setRunState) + recordStepEnd('done') stepsDone++ const jump = await handleLoopEnd(i) @@ -972,6 +1030,7 @@ export const useWorkflowRunStore = create((set, get) => { } catch (err) { if (!_cancel.current) { set((s) => ({ runState: { ...s.runState, status: 'error', error: String(err) }, activeNodeId: null })) + finishRecord('error', { error: String(err) }) useAppStore.getState().updateCurrentJob({ status: 'error', error: String(err) }) void showRunErrorNotification({ title: 'Workflow run failed', name: workflow.name, error: err }) } @@ -1000,6 +1059,7 @@ export const useWorkflowRunStore = create((set, get) => { const descendants = descendantWaits(waitId, ctx) _cancel.current = false + updateRecord(reopenWorkflowRunRecord) // Reset outputs for this branch's nodes so Retry re-executes cleanly. for (const node of branch) ctx.nodeOutputs.delete(node.id) @@ -1040,6 +1100,7 @@ export const useWorkflowRunStore = create((set, get) => { if (allFinished && !anyError) { finalize(ctx, newWaitStates) } else { + if (allFinished) finishRecord('error', { error: err ?? get().runState.error }) set((s) => ({ activeNodeId: null, runningBranchId: null, @@ -1062,15 +1123,19 @@ export const useWorkflowRunStore = create((set, get) => { activeNodeId: node.id, runState: { ...s.runState, blockIndex: i, blockProgress: 0, blockStep: 'Starting…' }, })) + recordStepStart(node, ctx.allExtensions) await executeExtensionNode(node, ctx, setRunState) + recordStepEnd('done') } finishBranch('done') } catch (err) { + if (!_cancel.current) recordStepEnd('error') finishBranch('error', String(err)) } }, cancel() { + finishRecord('cancelled') _cancel.current = true _pauseRequested.current = false flushResume() // unblock a manual While pause so the run can tear down @@ -1085,6 +1150,7 @@ export const useWorkflowRunStore = create((set, get) => { }, reset() { + finishRecord('cancelled') _ctx.current = null set({ runState: IDLE, activeNodeId: null, activeWorkflowId: null, nodeImageOutputs: {}, waitStates: {}, runningBranchId: null, whileProgress: {}, pausedGroup: [] }) }, diff --git a/src/shared/types/runs.ts b/src/shared/types/runs.ts new file mode 100644 index 000000000..c8fda5b63 --- /dev/null +++ b/src/shared/types/runs.ts @@ -0,0 +1,44 @@ +// Persisted run history -- mirrors api/schemas/runs.py (RunRecord). One record per +// generation, workflow or mesh-op run, listed by GET /workflow-runs. This is the +// shape the provenance model (T-F1) is meant to reuse rather than a second store. +// Timestamps are epoch seconds; output_path is workspace-relative when the output +// lives in the workspace, absolute otherwise. + +export type RunKind = 'generation' | 'workflow' | 'mesh_op' +/** 'interrupted' = still in flight when the backend stopped; terminal. */ +export type RunStatus = 'pending' | 'running' | 'done' | 'error' | 'cancelled' | 'interrupted' + +export interface RunStep { + name: string + node_id?: string | null + status: RunStatus + started_at: number + ended_at?: number | null +} + +export interface RunRecord { + id: string + kind: RunKind + status: RunStatus + model_id?: string | null + workflow_id?: string | null + workflow_name?: string | null + op_id?: string | null + inputs: Record + params: Record + steps: RunStep[] + error?: string | null + output_path?: string | null + created_at: number + started_at?: number | null + ended_at?: number | null + /** Computed by the backend; null while the run has not ended. */ + duration_s?: number | null +} + +export interface RunRecordPage { + runs: RunRecord[] + total: number + limit: number + offset: number +}