public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-18 21:47:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(slicer): Pipelines — multi-copy + class targeting + fanout + runs dashboard + retry-failed + WS updates (#1425 PR C — completes the v3 design)

Browse Source 
PR A/B turned the slice modal's preset bundle into a one-click dispatch
with a pinned target printer. PR C closes the original issue: operators
type in a number of copies, Bambuddy slices once and distributes prints
across a fleet per the pipeline's chosen fanout strategy. A new dashboard
surfaces every run with filters, expandable per-copy status, cancel,
and retry-failed-copies. WS pushes keep everything live.

Backend
- copies field on POST /run, capped by new pipeline_max_copies setting
  (default 50, hard cap 1000). PipelineRun.parent_run_id chains retries.
- SlicerPipelineUpdate accepts target_kind (specific_printer /
  printer_class), target_model_class, fanout_strategy.
- Eligibility matcher branches: class-targeting enumerates matching
  Printer rows, runs per-printer checks via a status_lookup closure,
  returns printer_reports[]. New issue kinds: no_class_matches,
  class_not_set.
- _pick_assignments distributes copies per strategy:
  - max_parallel: target_model set, printer_id None — scheduler picks
  - round_robin: copy i → eligible[i % N], fixed printer_id
  - fill_one_first: all copies pinned to eligible[0]
  All three reuse the slice-once path through slice_dispatch.enqueue.
- New routes:
  - GET /pipeline-runs (paginated, filterable by pipeline + status)
  - POST /pipeline-runs/{id}/retry-failed (creates child run with
    copies = failed+cancelled count, parent_run_id set)
  - Cancel cascades to all N queue entries (only pending/queued)
- _roll_up_run_status computes run-level status from per-job statuses;
  introduces partial_failure for "some completed, some failed".
- ws_manager.broadcast_to_user emits pipeline_run_updated on every
  state transition with the full materialised response.

Frontend
- Pipeline editor: target_kind radio + class picker (filtered to
  installed models) + fanout-strategy radio. Read-only row shows
  "X1C · Round robin" for class pipelines.
- RunWithPipelineModal: copies number input bounded by
  settings.pipeline_max_copies. Accepts class-targeted pipelines.
- Settings → Workflow → Queue & Dispatch: new "Slicer Pipeline limits"
  card with the max-copies input.
- New /pipelines/runs dashboard page (sidebar entry, gated on
  pipelines:read). Two-filter dropdown, 25-per-page pagination, per-row
  expandable to job list, Cancel + Retry-failed buttons.
- useWebSocket case for pipeline_run_updated invalidates both
  pipeline-runs-all and pipeline-runs/{id} query keys.
This commit is contained in:
maziggy
2026-06-27 16:52:05 +02:00
parent 4bbf0f031e
commit 3ef197e4e0
37 changed files with 2476 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/pipeline_runs.py
+369 -132
View File
@@ -1,18 +1,24 @@
"""API routes for Slicer Pipeline runs (#1425 PR B).
"""API routes for Slicer Pipeline runs (#1425 PR B + PR C).
PR B implements single-target dispatch: one Run-pipeline click =
slice the source file once with the pipeline's four preset slots →
enqueue one print on the pipeline's pinned target_printer_id.
PR B implemented single-target dispatch: one Run-pipeline click =
slice the source once → enqueue ONE print on ``target_printer_id``.
PR C extends this with copies > 1 + class targeting + fanout strategies;
the data model already carries those columns so this file is the only
place that changes shape.
PR C extends this with:
* ``copies > 1`` — slice once, enqueue N copies.
* ``target_kind='printer_class'`` — pipeline targets a Bambu model code
(X1C / P1S / H2D / …); orchestrator distributes copies across matching
printers using the pipeline's ``fanout_strategy``.
* Retry-failed runs that re-attempt only the failed/cancelled copies of
a partial-failure run.
* Dashboard list endpoint (``GET /pipeline-runs``) with status + pipeline
filters and pagination.
* WebSocket ``pipeline_run_updated`` events on state transitions so the
dashboard refreshes live without polling.
The slice is enqueued via ``slice_dispatch`` (the same path the manual
SliceModal uses), so the in-process progress toast renders for pipeline
runs the same way it does for manual slicing. The slice job's id lives
on PipelineRun.slice_job_id and is returned from ``POST /run`` so the
frontend can call ``trackJob`` directly.
The slice itself runs through ``slice_dispatch`` (same path as the manual
SliceModal), so the ``Slicing X — Generating G-code 75%`` toast renders
end-to-end. The slice job's id rides on the run response so the frontend
can call ``trackJob`` directly.
"""
from __future__ import annotations
@@ -24,13 +30,14 @@ from pathlib import Path
from typing import Literal
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy import desc, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.core.auth import RequirePermissionIfAuthEnabled
from backend.app.core.config import settings as app_settings
from backend.app.core.database import async_session, get_db
from backend.app.core.permissions import Permission
from backend.app.core.websocket import ws_manager
from backend.app.models.archive import PrintArchive
from backend.app.models.library import LibraryFile
from backend.app.models.pipeline_run import PipelineJob, PipelineRun
@@ -42,6 +49,7 @@ from backend.app.schemas.pipeline_run import (
CheckEligibilityRequest,
EligibilityIssueResponse,
EligibilityReportResponse,
PerPrinterReport as PerPrinterReportResponse,
PipelineJobResponse,
PipelineRunCreateRequest,
PipelineRunListResponse,
@@ -56,9 +64,6 @@ from backend.app.services.pipeline_eligibility import (
logger = logging.getLogger(__name__)
# Two routers — one for the per-pipeline endpoints (mounted under the
# existing ``/slicer-pipelines`` prefix) and one for the per-run endpoints
# at ``/pipeline-runs``. Keeps the URL shape natural.
pipeline_run_create_router = APIRouter(prefix="/slicer-pipelines", tags=["Slicer Pipelines"])
pipeline_run_router = APIRouter(prefix="/pipeline-runs", tags=["Slicer Pipelines"])
@@ -71,8 +76,10 @@ pipeline_run_router = APIRouter(prefix="/pipeline-runs", tags=["Slicer Pipelines
def _serialise_status(report: EligibilityReport) -> EligibilityReportResponse:
return EligibilityReportResponse(
ok=report.ok,
target_kind=report.target_kind,
target_printer_id=report.target_printer_id,
target_printer_name=report.target_printer_name,
target_model_class=report.target_model_class,
issues=[
EligibilityIssueResponse(
kind=issue.kind,
@@ -82,6 +89,23 @@ def _serialise_status(report: EligibilityReport) -> EligibilityReportResponse:
)
for issue in report.issues
],
printer_reports=[
PerPrinterReportResponse(
printer_id=r.printer_id,
printer_name=r.printer_name,
ok=r.ok,
issues=[
EligibilityIssueResponse(
kind=i.kind,
slot_index=i.slot_index,
expected=i.expected,
actual=i.actual,
)
for i in r.issues
],
)
for r in report.printer_reports
],
)
@@ -101,8 +125,7 @@ async def _load_pipeline(db: AsyncSession, pipeline_id: int) -> SlicerPipeline:
async def _load_printer_status(printer_id: int | None) -> dict | None:
"""Snapshot the printer_manager's live PrinterState for the eligibility
matcher. Returns ``None`` when the printer has no MQTT client (offline at
the manager level)."""
matcher. Returns ``None`` when the printer has no MQTT client."""
if printer_id is None:
return None
from backend.app.services.printer_manager import printer_manager
@@ -113,9 +136,23 @@ async def _load_printer_status(printer_id: int | None) -> dict | None:
return {"connected": state.connected, "raw_data": state.raw_data}
def _make_status_lookup():
"""Closure that snapshots the printer_manager once per printer_id call.
Passed to the matcher's class-targeting branch so it can read live state
for every candidate printer."""
def _lookup(printer_id: int) -> dict | None:
from backend.app.services.printer_manager import printer_manager
state = printer_manager.get_status(printer_id)
if state is None:
return None
return {"connected": state.connected, "raw_data": state.raw_data}
return _lookup
def _slice_request_from_pipeline(pipeline: SlicerPipeline) -> SliceRequest:
"""Materialise a SliceRequest from a pipeline so we can hand it to
``slice_and_persist`` exactly the way the existing SliceModal flow does."""
try:
raw_filaments = json.loads(pipeline.filament_presets_json or "[]")
except (json.JSONDecodeError, TypeError):
@@ -134,25 +171,6 @@ def _slice_request_from_pipeline(pipeline: SlicerPipeline) -> SliceRequest:
)
def _compute_run_status(
persisted: str,
job: PipelineJob | None,
queue_entry: PrintQueueItem | None,
) -> str:
if persisted in ("failed", "cancelled", "completed"):
return persisted
if queue_entry is None:
return persisted
queue_status = queue_entry.status
if queue_status == "completed":
return "completed"
if queue_status in ("failed", "cancelled", "aborted"):
return "failed" if queue_status != "cancelled" else "cancelled"
if queue_status == "printing":
return "in_progress"
return "dispatching" if persisted == "dispatching" else "queued"
def _compute_job_status(
persisted: str,
queue_entry: PrintQueueItem | None,
@@ -173,13 +191,61 @@ def _compute_job_status(
return "queued"
def _roll_up_run_status(
persisted: str,
job_statuses: list[str],
) -> str:
"""Compute the run-level status from the per-job statuses.
Terminal-persisted always wins for explicit cancels / hard failures so
the dashboard doesn't flicker when one job's queue entry hasn't caught
up. Otherwise:
- all completed → completed
- any in_progress / printing / queued / dispatching → in_progress
- any failed alongside any completed → partial_failure
- all failed/cancelled → failed
"""
if persisted in ("cancelled",):
return persisted
if not job_statuses:
return persisted
completed = sum(1 for s in job_statuses if s == "completed")
failed = sum(1 for s in job_statuses if s == "failed")
cancelled = sum(1 for s in job_statuses if s == "cancelled")
in_flight = sum(1 for s in job_statuses if s in ("printing", "queued", "awaiting_printer", "pending"))
total = len(job_statuses)
if completed == total:
return "completed"
if in_flight > 0:
return "in_progress" if persisted not in ("queued", "slicing", "dispatching") else persisted
# All copies are in terminal states.
if failed == 0 and cancelled == total:
return "cancelled"
if completed > 0 and (failed > 0 or cancelled > 0):
return "partial_failure"
if failed > 0:
return "failed"
return persisted
async def _materialise_run(db: AsyncSession, run: PipelineRun) -> PipelineRunResponse:
pipeline_name: str | None = None
target_kind = None
target_printer_id = None
target_model_class = None
fanout_strategy = None
if run.pipeline_id:
pipeline = (
await db.execute(select(SlicerPipeline).where(SlicerPipeline.id == run.pipeline_id))
).scalar_one_or_none()
pipeline_name = pipeline.name if pipeline else None
if pipeline:
pipeline_name = pipeline.name
target_kind = pipeline.target_kind # type: ignore[assignment]
target_printer_id = pipeline.target_printer_id
target_model_class = pipeline.target_model_class
fanout_strategy = pipeline.fanout_strategy # type: ignore[assignment]
source_filename: str | None = None
if run.source_library_file_id:
@@ -204,7 +270,7 @@ async def _materialise_run(db: AsyncSession, run: PipelineRun) -> PipelineRunRes
)
job_responses: list[PipelineJobResponse] = []
rolled_up_status = run.status
job_live_statuses: list[str] = []
for job in job_rows:
queue_entry = None
if job.queue_entry_id:
@@ -218,6 +284,7 @@ async def _materialise_run(db: AsyncSession, run: PipelineRun) -> PipelineRunRes
printer_name = p.name if p else None
live_job_status = _compute_job_status(job.status, queue_entry)
job_live_statuses.append(live_job_status)
job_responses.append(
PipelineJobResponse(
id=job.id,
@@ -232,7 +299,8 @@ async def _materialise_run(db: AsyncSession, run: PipelineRun) -> PipelineRunRes
completed_at=job.completed_at,
)
)
rolled_up_status = _compute_run_status(run.status, job, queue_entry)
rolled_up = _roll_up_run_status(run.status, job_live_statuses)
return PipelineRunResponse(
id=run.id,
@@ -241,8 +309,15 @@ async def _materialise_run(db: AsyncSession, run: PipelineRun) -> PipelineRunRes
source_library_file_id=run.source_library_file_id,
source_archive_id=run.source_archive_id,
source_filename=source_filename,
parent_run_id=run.parent_run_id,
copies=run.copies,
status=rolled_up_status, # type: ignore[arg-type]
copies_completed=sum(1 for s in job_live_statuses if s == "completed"),
copies_failed=sum(1 for s in job_live_statuses if s == "failed"),
copies_cancelled=sum(1 for s in job_live_statuses if s == "cancelled"),
copies_in_progress=sum(
1 for s in job_live_statuses if s in ("printing", "queued", "awaiting_printer", "pending")
),
status=rolled_up, # type: ignore[arg-type]
slice_job_id=run.slice_job_id,
sliced_library_file_id=run.sliced_library_file_id,
eligibility_overridden=run.eligibility_overridden,
@@ -252,9 +327,30 @@ async def _materialise_run(db: AsyncSession, run: PipelineRun) -> PipelineRunRes
started_at=run.started_at,
completed_at=run.completed_at,
jobs=job_responses,
target_kind=target_kind,
target_printer_id=target_printer_id,
target_model_class=target_model_class,
fanout_strategy=fanout_strategy,
)
async def _publish_run_event(db: AsyncSession, run: PipelineRun) -> None:
"""Broadcast a ``pipeline_run_updated`` event with the full materialised
run. Per-user routing via ``broadcast_to_user`` falls back to a global
broadcast when ``created_by`` is None (auth-disabled installs)."""
try:
payload = await _materialise_run(db, run)
await ws_manager.broadcast_to_user(
run.created_by,
{
"type": "pipeline_run_updated",
"run": payload.model_dump(mode="json"),
},
)
except Exception:
logger.exception("Failed to broadcast pipeline_run_updated for run %d", run.id)
# ---------------------------------------------------------------------------
# Source resolution + orchestration
# ---------------------------------------------------------------------------
@@ -269,16 +365,13 @@ async def _resolve_source(
library_file_id: int | None,
archive_id: int | None,
) -> tuple[SourceKind, int, str, Path]:
"""Look up the source row + on-disk path for either input kind. Raises 404
when the row or its file is missing — same shape as the SliceModal's flow
for both libraryFile and archive."""
if library_file_id is not None:
lib = (await db.execute(select(LibraryFile).where(LibraryFile.id == library_file_id))).scalar_one_or_none()
if lib is None:
raise HTTPException(404, "Source library file not found")
src_path = (
Path(app_settings.base_dir) / lib.file_path
) # SEC-PATH-OK: lib.file_path is a LibraryFile DB column set only by the upload route (routes/library.py POST /files), which writes a UUID-named file under base_dir/library_files/ — never user-controlled at this site.
) # SEC-PATH-OK: lib.file_path is a LibraryFile DB column set only by the upload route, which writes a UUID-named file under base_dir/library_files/.
if not src_path.exists():
raise HTTPException(404, "Source library file missing on disk")
return ("library_file", lib.id, lib.filename, src_path)
@@ -292,13 +385,60 @@ async def _resolve_source(
raise HTTPException(400, "Archive has no source file to slice")
src_path = (
Path(app_settings.base_dir) / rel
) # SEC-PATH-OK: rel is archive.source_3mf_path / archive.file_path, both set by upload-time validators (_resolve_source_3mf_path + archive ingestion) that already do resolve+relative_to containment. Mirrors routes/archives.py:3955.
) # SEC-PATH-OK: rel is archive.source_3mf_path / archive.file_path, both set by upload-time validators that already do resolve+relative_to containment.
if not src_path.exists():
raise HTTPException(404, "Archive source file missing on disk")
name = arc.filename or arc.print_name or src_path.name
return ("archive", arc.id, name, src_path)
async def _pick_assignments(
db: AsyncSession,
pipeline: SlicerPipeline,
copies: int,
) -> list[tuple[int | None, str | None]]:
"""Return ``[(printer_id_or_None, target_model_or_None), ...]`` of length
``copies`` per the pipeline's fanout strategy. ``target_model_class``
items leave ``printer_id`` None so the scheduler picks any free matching
printer; specific assignments fill ``printer_id``."""
target_kind = pipeline.target_kind or "specific_printer"
if target_kind == "specific_printer" or pipeline.target_printer_id is not None:
assert pipeline.target_printer_id is not None
return [(pipeline.target_printer_id, None)] * copies
# Class-targeting. Enumerate matching printers + apply the strategy.
matching = (
(
await db.execute(
select(Printer)
.where(Printer.model == pipeline.target_model_class)
.where(Printer.is_active.is_(True))
.order_by(Printer.id)
)
)
.scalars()
.all()
)
if not matching:
# Shouldn't reach here when eligibility passes, but failing gracefully
# is better than a TypeError on next-slot pick.
return [(None, pipeline.target_model_class)] * copies
strategy = pipeline.fanout_strategy or "max_parallel"
if strategy == "fill_one_first":
# Pin every copy to the first match. Scheduler dispatches them serially
# to that printer. If the printer breaks, copies wait; that's the
# documented trade-off.
return [(matching[0].id, None)] * copies
if strategy == "round_robin":
# Cycle through eligible printers — copy ``i`` lands on
# ``matching[i % len(matching)]``. Each item gets a fixed printer_id.
return [(matching[i % len(matching)].id, None) for i in range(copies)]
# max_parallel — leave printer_id=None, set target_model so the scheduler
# picks any free X1C / P1S / … for each item independently.
return [(None, pipeline.target_model_class)] * copies
def _make_orchestration_callable(
*,
run_id: int,
@@ -307,18 +447,13 @@ def _make_orchestration_callable(
src_id: int,
src_filename: str,
src_path: Path,
target_printer_id: int,
creator_user_id: int | None,
copies: int,
):
"""Return the async callable that ``slice_dispatch.enqueue`` will run as
the background slice job. Wrapping it as the slice job's ``run`` means the
SliceJob's lifecycle (pending → running → completed/failed) drives the
progress toast on the frontend exactly the same as a manual SliceModal
slice — no separate notification surface for pipeline runs."""
"""Returns the async callable that ``slice_dispatch.enqueue`` runs as the
background slice job. Wraps slice + multi-copy enqueue + state update."""
async def _orchestrate(slice_job_id: int) -> dict:
# Local import — slice_and_persist lives in routes/library which
# imports back into this module transitively via slicer_pipelines.
from backend.app.api.routes.library import slice_and_persist
async with async_session() as session:
@@ -327,25 +462,17 @@ def _make_orchestration_callable(
await session.execute(select(SlicerPipeline).where(SlicerPipeline.id == pipeline_id))
).scalar_one_or_none()
if run is None or pipeline is None:
logger.warning(
"pipeline_run %d or pipeline %d disappeared mid-orchestration",
run_id,
pipeline_id,
)
logger.warning("pipeline_run %d or pipeline %d disappeared mid-orchestration", run_id, pipeline_id)
return {}
# Refresh the snapshot status now that slicing has actually
# started — the route handler already wrote slice_job_id but
# left status='queued' until this point.
run.status = "slicing"
run.started_at = datetime.now(timezone.utc)
await session.commit()
await _publish_run_event(session, run)
slice_request = _slice_request_from_pipeline(pipeline)
model_bytes = src_path.read_bytes()
# Resolve the folder for the sliced output. Library sources
# keep their folder; archive sources fall through to root.
folder_id: int | None = None
if src_kind == "library_file":
lib = (await session.execute(select(LibraryFile).where(LibraryFile.id == src_id))).scalar_one_or_none()
@@ -365,17 +492,14 @@ def _make_orchestration_callable(
},
request=slice_request,
current_user_id=creator_user_id,
job_id=slice_job_id, # threads --pipe progress
job_id=slice_job_id,
)
except HTTPException as exc:
# _SliceJobError isn't directly importable here without a
# heavier dep; surface the slice failure cleanly and let
# the dispatcher's generic Exception path mark the
# SliceJob failed. Persist the pipeline-side state here.
run.status = "failed"
run.error_message = f"Slice failed: {exc.detail}"
run.completed_at = datetime.now(timezone.utc)
await session.commit()
await _publish_run_event(session, run)
raise
except Exception as exc:
logger.exception("Pipeline run %d slice raised unexpectedly", run_id)
@@ -383,40 +507,47 @@ def _make_orchestration_callable(
run.error_message = f"Slice failed: {exc}"
run.completed_at = datetime.now(timezone.utc)
await session.commit()
await _publish_run_event(session, run)
raise
run.sliced_library_file_id = slice_response.library_file_id
# PR B copies=1 — exactly one PipelineJob to dispatch.
job = (
(await session.execute(select(PipelineJob).where(PipelineJob.pipeline_run_id == run_id)))
# PR C: enqueue N copies per the picked assignment strategy.
assignments = await _pick_assignments(session, pipeline, copies)
jobs = (
(
await session.execute(
select(PipelineJob)
.where(PipelineJob.pipeline_run_id == run_id)
.order_by(PipelineJob.copy_index)
)
)
.scalars()
.first()
.all()
)
if job is None:
logger.warning("pipeline_run %d has no PipelineJob row", run_id)
run.status = "failed"
run.error_message = "Internal: missing pipeline_job row"
run.completed_at = datetime.now(timezone.utc)
await session.commit()
return slice_response.model_dump()
if len(jobs) != copies:
logger.warning("pipeline_run %d expected %d jobs, found %d", run_id, copies, len(jobs))
queue_item = PrintQueueItem(
printer_id=target_printer_id,
library_file_id=slice_response.library_file_id,
created_by_id=creator_user_id,
status="pending",
)
session.add(queue_item)
await session.flush()
for job, (printer_id, target_model) in zip(jobs, assignments, strict=False):
queue_item = PrintQueueItem(
printer_id=printer_id,
target_model=target_model,
library_file_id=slice_response.library_file_id,
created_by_id=creator_user_id,
status="pending",
)
session.add(queue_item)
await session.flush()
job.queue_entry_id = queue_item.id
job.assigned_printer_id = target_printer_id
job.status = "queued"
job.dispatched_at = datetime.now(timezone.utc)
job.queue_entry_id = queue_item.id
job.assigned_printer_id = printer_id # may be None for max_parallel
job.status = "queued"
job.dispatched_at = datetime.now(timezone.utc)
run.status = "dispatching"
await session.commit()
await _publish_run_event(session, run)
return slice_response.model_dump()
@@ -435,19 +566,17 @@ async def check_eligibility(
_: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_READ),
db: AsyncSession = Depends(get_db),
):
"""Pre-flight check before a Run-pipeline click. Returns an eligibility
report; the frontend uses it to show a confirmation modal if any blocking
issue exists. Accepts either a library file id or an archive id (XOR)."""
pipeline = await _load_pipeline(db, pipeline_id)
# Source resolution — same shape as /run uses, so a 404 here means /run
# would also 404 and the user can't proceed.
await _resolve_source(
db,
library_file_id=body.source_library_file_id,
archive_id=body.source_archive_id,
)
status = await _load_printer_status(pipeline.target_printer_id)
report = await check_pipeline_eligibility(db, pipeline, status)
if pipeline.target_kind == "printer_class" and pipeline.target_printer_id is None:
report = await check_pipeline_eligibility(db, pipeline, status_lookup=_make_status_lookup())
else:
status = await _load_printer_status(pipeline.target_printer_id)
report = await check_pipeline_eligibility(db, pipeline, status)
return _serialise_status(report)
@@ -463,10 +592,7 @@ async def run_pipeline(
current_user: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_RUN),
db: AsyncSession = Depends(get_db),
):
"""Kick off a pipeline run. PR B: copies=1, target = pipeline's pinned
printer. Returns 202 immediately; the slice runs in the background via
``slice_dispatch`` and the slice job id rides on the response so the
frontend can attach the progress toast."""
from backend.app.api.routes.settings import get_setting
from backend.app.services.slice_dispatch import slice_dispatch
pipeline = await _load_pipeline(db, pipeline_id)
@@ -476,24 +602,40 @@ async def run_pipeline(
archive_id=body.source_archive_id,
)
# Eligibility pre-flight. ``force=True`` from the confirmation modal
# bypasses the 409.
status = await _load_printer_status(pipeline.target_printer_id)
report = await check_pipeline_eligibility(db, pipeline, status)
# Cap copies against the configured ceiling.
raw_cap = await get_setting(db, "pipeline_max_copies")
try:
cap = int(raw_cap) if raw_cap else 50
except (TypeError, ValueError):
cap = 50
if body.copies > cap:
raise HTTPException(
422,
f"copies={body.copies} exceeds pipeline_max_copies setting ({cap})",
)
# Eligibility pre-flight.
if pipeline.target_kind == "printer_class" and pipeline.target_printer_id is None:
report = await check_pipeline_eligibility(db, pipeline, status_lookup=_make_status_lookup())
else:
status = await _load_printer_status(pipeline.target_printer_id)
report = await check_pipeline_eligibility(db, pipeline, status)
if not report.ok and not body.force:
raise HTTPException(status_code=409, detail=_serialise_status(report).model_dump())
if pipeline.target_printer_id is None:
# Need a target — specific or class — to dispatch.
if pipeline.target_printer_id is None and not pipeline.target_model_class:
raise HTTPException(
400,
"Pipeline has no target printer set. Open the pipeline in Settings → Workflow → Pipelines and choose a target.",
"Pipeline has no target. Open the pipeline in Settings → Workflow → Pipelines and choose a target printer or printer class.",
)
run = PipelineRun(
pipeline_id=pipeline.id,
source_library_file_id=src_id if src_kind == "library_file" else None,
source_archive_id=src_id if src_kind == "archive" else None,
copies=1,
copies=body.copies,
status="queued",
eligibility_overridden=(not report.ok and body.force),
created_by=current_user.id if current_user else None,
@@ -501,19 +643,19 @@ async def run_pipeline(
db.add(run)
await db.flush()
job = PipelineJob(
pipeline_run_id=run.id,
copy_index=0,
assigned_printer_id=pipeline.target_printer_id,
status="pending",
)
db.add(job)
# One PipelineJob per copy. PR B was copies=1, PR C generalises.
for i in range(body.copies):
db.add(
PipelineJob(
pipeline_run_id=run.id,
copy_index=i,
status="pending",
)
)
await db.commit()
await db.refresh(run)
await _publish_run_event(db, run)
# Enqueue the slice job. The callable inside slice_dispatch.enqueue does
# the full slice → enqueue-print → state-update chain. SliceJob lifecycle
# drives the existing progress toast.
orchestrate = _make_orchestration_callable(
run_id=run.id,
pipeline_id=pipeline.id,
@@ -521,8 +663,8 @@ async def run_pipeline(
src_id=src_id,
src_filename=src_filename,
src_path=src_path,
target_printer_id=pipeline.target_printer_id,
creator_user_id=current_user.id if current_user else None,
copies=body.copies,
)
slice_job = await slice_dispatch.enqueue(
kind="library_file" if src_kind == "library_file" else "archive",
@@ -539,7 +681,7 @@ async def run_pipeline(
# ---------------------------------------------------------------------------
# /slicer-pipelines/{id}/runs + /pipeline-runs/{id} + cancel
# Lists, reads, cancel, retry-failed
# ---------------------------------------------------------------------------
@@ -563,7 +705,47 @@ async def list_runs_for_pipeline(
.scalars()
.all()
)
return PipelineRunListResponse(runs=[await _materialise_run(db, r) for r in rows])
total = (
await db.execute(select(func.count()).select_from(PipelineRun).where(PipelineRun.pipeline_id == pipeline_id))
).scalar() or 0
return PipelineRunListResponse(
runs=[await _materialise_run(db, r) for r in rows],
total=total,
)
@pipeline_run_router.get("", response_model=PipelineRunListResponse)
async def list_all_runs(
limit: int = 25,
offset: int = 0,
pipeline_id: int | None = None,
status: str | None = None,
_: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_READ),
db: AsyncSession = Depends(get_db),
):
"""Dashboard list. Newest first; filters on pipeline_id + status. The
`status` filter matches the persisted snapshot, not the live roll-up —
in-progress runs may appear under `dispatching` until the next state
transition writes through."""
limit = max(1, min(limit, 100))
offset = max(0, offset)
stmt = select(PipelineRun)
count_stmt = select(func.count()).select_from(PipelineRun)
if pipeline_id is not None:
stmt = stmt.where(PipelineRun.pipeline_id == pipeline_id)
count_stmt = count_stmt.where(PipelineRun.pipeline_id == pipeline_id)
if status:
stmt = stmt.where(PipelineRun.status == status)
count_stmt = count_stmt.where(PipelineRun.status == status)
rows = (await db.execute(stmt.order_by(desc(PipelineRun.id)).offset(offset).limit(limit))).scalars().all()
total = (await db.execute(count_stmt)).scalar() or 0
return PipelineRunListResponse(
runs=[await _materialise_run(db, r) for r in rows],
total=total,
)
@pipeline_run_router.get("/{run_id}", response_model=PipelineRunResponse)
@@ -584,13 +766,13 @@ async def cancel_run(
_: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_RUN),
db: AsyncSession = Depends(get_db),
):
"""Cancel a queued / in-flight run. Pre-slice or post-slice — semantics
are described on the original PR B route."""
"""Cancel a queued / in-flight run. Cascades to all non-terminal queue
entries; in-flight prints continue on the printer (operator must Stop)."""
run = (await db.execute(select(PipelineRun).where(PipelineRun.id == run_id))).scalar_one_or_none()
if run is None:
raise HTTPException(404, "Pipeline run not found")
if run.status in ("completed", "failed", "cancelled"):
if run.status in ("completed", "failed", "cancelled", "partial_failure"):
return await _materialise_run(db, run)
run.status = "cancelled"
@@ -604,10 +786,7 @@ async def cancel_run(
queue_entry = (
await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == job.queue_entry_id))
).scalar_one_or_none()
if queue_entry is not None and queue_entry.status in (
"pending",
"queued",
):
if queue_entry is not None and queue_entry.status in ("pending", "queued"):
queue_entry.status = "cancelled"
if job.status not in ("completed", "failed", "cancelled"):
job.status = "cancelled"
@@ -615,4 +794,62 @@ async def cancel_run(
await db.commit()
await db.refresh(run)
await _publish_run_event(db, run)
return await _materialise_run(db, run)
@pipeline_run_router.post("/{run_id}/retry-failed", response_model=PipelineRunResponse, status_code=202)
async def retry_failed(
run_id: int,
current_user: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_RUN),
db: AsyncSession = Depends(get_db),
):
"""Create a new run with copies = (failed + cancelled count) from the
parent. Same pipeline, same source. Eligibility re-checked at run time
(it might pass this time — operator may have fixed the issue)."""
parent = (await db.execute(select(PipelineRun).where(PipelineRun.id == run_id))).scalar_one_or_none()
if parent is None:
raise HTTPException(404, "Pipeline run not found")
if parent.pipeline_id is None:
raise HTTPException(400, "Original pipeline was deleted; cannot retry")
if parent.source_library_file_id is None and parent.source_archive_id is None:
raise HTTPException(400, "Original source was deleted; cannot retry")
# Count the parent's failed + cancelled jobs.
parent_jobs = (
(await db.execute(select(PipelineJob).where(PipelineJob.pipeline_run_id == parent.id))).scalars().all()
)
fail_count = 0
for j in parent_jobs:
queue_entry = None
if j.queue_entry_id:
queue_entry = (
await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == j.queue_entry_id))
).scalar_one_or_none()
live = _compute_job_status(j.status, queue_entry)
if live in ("failed", "cancelled"):
fail_count += 1
if fail_count == 0:
raise HTTPException(400, "No failed copies to retry")
# Build the request payload the same way the user would have via /run.
body = PipelineRunCreateRequest(
source_library_file_id=parent.source_library_file_id,
source_archive_id=parent.source_archive_id,
copies=fail_count,
force=True, # operator already accepted eligibility on the parent
)
# Reuse the run_pipeline route logic via a direct call — keeps the
# orchestration single-sourced. The result inherits parent_run_id.
new_run_response = await run_pipeline(parent.pipeline_id, body, current_user=current_user, db=db)
# Stamp parent_run_id on the freshly-created run.
new_row = (await db.execute(select(PipelineRun).where(PipelineRun.id == new_run_response.id))).scalar_one_or_none()
if new_row is not None:
new_row.parent_run_id = parent.id
await db.commit()
await db.refresh(new_row)
return await _materialise_run(db, new_row)
return new_run_response
backend/app/api/routes/settings.py
+1
View File
@@ -164,6 +164,7 @@ async def _build_settings_response(db: AsyncSession, is_api_key: bool = False) -
"stagger_interval_minutes",
"forecast_global_lead_time_days",
"session_max_hours",
"pipeline_max_copies",
]:
settings_dict[setting.key] = int(setting.value)
elif setting.key == "default_printer_id":
backend/app/api/routes/slicer_pipelines.py
+6
View File
@@ -166,6 +166,12 @@ async def update_pipeline(
row.target_printer_id = None
else:
row.target_printer_id = data.target_printer_id
# PR C — class targeting + fanout strategy. Empty string from the frontend
# also clears the class (radio toggled away).
if data.target_model_class is not None:
row.target_model_class = data.target_model_class or None
if data.fanout_strategy is not None:
row.fanout_strategy = data.fanout_strategy
await db.commit()
await db.refresh(row)