Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion api/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
37 changes: 36 additions & 1 deletion api/routers/generation.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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
)
Expand All @@ -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()
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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(
Expand All @@ -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
Expand All @@ -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
Expand Down
42 changes: 40 additions & 2 deletions api/routers/optimize.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import hashlib
import os
import tempfile
import time
import uuid
from contextlib import contextmanager

Expand All @@ -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"])

Expand Down Expand Up @@ -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]:
Expand Down
77 changes: 73 additions & 4 deletions api/routers/workflow_runs.py
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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"])

Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand Down
62 changes: 62 additions & 0 deletions api/schemas/runs.py
Original file line number Diff line number Diff line change
@@ -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
Loading
Loading