bambuddy/backend/app/api/routes/pipeline_runs.py

"""API routes for Slicer Pipeline runs (#1425 PR B + PR C).

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`` — 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 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

import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal

from fastapi import APIRouter, Depends, HTTPException
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
from backend.app.models.print_queue import PrintQueueItem
from backend.app.models.printer import Printer
from backend.app.models.slicer_pipeline import SlicerPipeline
from backend.app.models.user import User
from backend.app.schemas.pipeline_run import (
    CheckEligibilityRequest,
    EligibilityIssueResponse,
    EligibilityReportResponse,
    PerPrinterReport as PerPrinterReportResponse,
    PipelineJobResponse,
    PipelineRunCreateRequest,
    PipelineRunListResponse,
    PipelineRunResponse,
)
from backend.app.schemas.slicer import PresetRef, SliceRequest
from backend.app.services.pipeline_eligibility import (
    EligibilityReport,
    check_pipeline_eligibility,
)

logger = logging.getLogger(__name__)


pipeline_run_create_router = APIRouter(prefix="/slicer-pipelines", tags=["Slicer Pipelines"])
pipeline_run_router = APIRouter(prefix="/pipeline-runs", tags=["Slicer Pipelines"])


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


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,
                slot_index=issue.slot_index,
                expected=issue.expected,
                actual=issue.actual,
            )
            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
        ],
    )


async def _load_pipeline(db: AsyncSession, pipeline_id: int) -> SlicerPipeline:
    pipeline = (
        await db.execute(
            select(SlicerPipeline).where(
                SlicerPipeline.id == pipeline_id,
                SlicerPipeline.is_deleted.is_(False),
            )
        )
    ).scalar_one_or_none()
    if pipeline is None:
        raise HTTPException(404, "Pipeline not found")
    return pipeline


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."""
    if printer_id is None:
        return 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}


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:
    try:
        raw_filaments = json.loads(pipeline.filament_presets_json or "[]")
    except (json.JSONDecodeError, TypeError):
        raw_filaments = []
    filament_presets = [
        PresetRef(source=r["source"], id=r["id"])
        for r in raw_filaments
        if isinstance(r, dict) and "source" in r and "id" in r
    ]
    return SliceRequest(
        printer_preset=PresetRef(source=pipeline.printer_preset_source, id=pipeline.printer_preset_id),
        process_preset=PresetRef(source=pipeline.process_preset_source, id=pipeline.process_preset_id),
        filament_presets=filament_presets,
        bed_type=pipeline.bed_type,
        export_3mf=True,
    )


def _compute_job_status(
    persisted: str,
    queue_entry: PrintQueueItem | None,
) -> str:
    if persisted in ("failed", "cancelled", "completed"):
        return persisted
    if queue_entry is None:
        return persisted
    qs = queue_entry.status
    if qs == "completed":
        return "completed"
    if qs in ("failed", "aborted"):
        return "failed"
    if qs == "cancelled":
        return "cancelled"
    if qs == "printing":
        return "printing"
    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()
        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:
        src = (
            await db.execute(select(LibraryFile).where(LibraryFile.id == run.source_library_file_id))
        ).scalar_one_or_none()
        source_filename = src.filename if src else None
    elif run.source_archive_id:
        arc = (
            await db.execute(select(PrintArchive).where(PrintArchive.id == run.source_archive_id))
        ).scalar_one_or_none()
        source_filename = (arc.print_name or arc.filename) if arc else None

    job_rows = (
        (
            await db.execute(
                select(PipelineJob).where(PipelineJob.pipeline_run_id == run.id).order_by(PipelineJob.copy_index)
            )
        )
        .scalars()
        .all()
    )

    job_responses: list[PipelineJobResponse] = []
    job_live_statuses: list[str] = []
    for job in job_rows:
        queue_entry = None
        if job.queue_entry_id:
            queue_entry = (
                await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == job.queue_entry_id))
            ).scalar_one_or_none()

        printer_name: str | None = None
        if job.assigned_printer_id:
            p = (await db.execute(select(Printer).where(Printer.id == job.assigned_printer_id))).scalar_one_or_none()
            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,
                pipeline_run_id=job.pipeline_run_id,
                copy_index=job.copy_index,
                assigned_printer_id=job.assigned_printer_id,
                assigned_printer_name=printer_name,
                queue_entry_id=job.queue_entry_id,
                status=live_job_status,  # type: ignore[arg-type]
                error_message=job.error_message,
                dispatched_at=job.dispatched_at,
                completed_at=job.completed_at,
            )
        )

    rolled_up = _roll_up_run_status(run.status, job_live_statuses)

    return PipelineRunResponse(
        id=run.id,
        pipeline_id=run.pipeline_id,
        pipeline_name=pipeline_name,
        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,
        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,
        error_message=run.error_message,
        created_by=run.created_by,
        created_at=run.created_at,
        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
# ---------------------------------------------------------------------------


SourceKind = Literal["library_file", "archive"]


async def _resolve_source(
    db: AsyncSession,
    *,
    library_file_id: int | None,
    archive_id: int | None,
) -> tuple[SourceKind, int, str, Path]:
    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, 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)

    assert archive_id is not None
    arc = (await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))).scalar_one_or_none()
    if arc is None:
        raise HTTPException(404, "Source archive not found")
    rel = arc.source_3mf_path or arc.file_path
    if not rel:
        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 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,
    pipeline_id: int,
    src_kind: SourceKind,
    src_id: int,
    src_filename: str,
    src_path: Path,
    creator_user_id: int | None,
    copies: int,
):
    """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:
        from backend.app.api.routes.library import slice_and_persist

        async with async_session() as session:
            run = (await session.execute(select(PipelineRun).where(PipelineRun.id == run_id))).scalar_one_or_none()
            pipeline = (
                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)
                return {}

            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()

            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()
                if lib is not None:
                    folder_id = lib.folder_id

            try:
                slice_response = await slice_and_persist(
                    session,
                    model_bytes=model_bytes,
                    model_filename=src_filename,
                    folder_id=folder_id,
                    extra_metadata={
                        f"sliced_from_{src_kind}_id": src_id,
                        "sliced_via_pipeline_id": pipeline.id,
                        "sliced_via_pipeline_run_id": run.id,
                    },
                    request=slice_request,
                    current_user_id=creator_user_id,
                    job_id=slice_job_id,
                )
            except HTTPException as exc:
                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)
                run.status = "failed"
                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 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()
                .all()
            )
            if len(jobs) != copies:
                logger.warning("pipeline_run %d expected %d jobs, found %d", run_id, copies, len(jobs))

            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 = 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()

    return _orchestrate


# ---------------------------------------------------------------------------
# /slicer-pipelines/{id}/check-eligibility
# ---------------------------------------------------------------------------


@pipeline_run_create_router.post("/{pipeline_id}/check-eligibility", response_model=EligibilityReportResponse)
async def check_eligibility(
    pipeline_id: int,
    body: CheckEligibilityRequest,
    _: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_READ),
    db: AsyncSession = Depends(get_db),
):
    pipeline = await _load_pipeline(db, pipeline_id)
    await _resolve_source(
        db,
        library_file_id=body.source_library_file_id,
        archive_id=body.source_archive_id,
    )
    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)


# ---------------------------------------------------------------------------
# /slicer-pipelines/{id}/run
# ---------------------------------------------------------------------------


@pipeline_run_create_router.post("/{pipeline_id}/run", response_model=PipelineRunResponse, status_code=202)
async def run_pipeline(
    pipeline_id: int,
    body: PipelineRunCreateRequest,
    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_RUN),
    db: AsyncSession = Depends(get_db),
):
    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)
    src_kind, src_id, src_filename, src_path = await _resolve_source(
        db,
        library_file_id=body.source_library_file_id,
        archive_id=body.source_archive_id,
    )

    # 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())

    # 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. 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=body.copies,
        status="queued",
        eligibility_overridden=(not report.ok and body.force),
        created_by=current_user.id if current_user else None,
    )
    db.add(run)
    await db.flush()

    # 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)

    orchestrate = _make_orchestration_callable(
        run_id=run.id,
        pipeline_id=pipeline.id,
        src_kind=src_kind,
        src_id=src_id,
        src_filename=src_filename,
        src_path=src_path,
        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",
        source_id=src_id,
        source_name=src_filename,
        run=orchestrate,
    )

    run.slice_job_id = slice_job.id
    await db.commit()
    await db.refresh(run)

    return await _materialise_run(db, run)


# ---------------------------------------------------------------------------
# Lists, reads, cancel, retry-failed
# ---------------------------------------------------------------------------


@pipeline_run_create_router.get("/{pipeline_id}/runs", response_model=PipelineRunListResponse)
async def list_runs_for_pipeline(
    pipeline_id: int,
    limit: int = 10,
    _: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_READ),
    db: AsyncSession = Depends(get_db),
):
    limit = max(1, min(limit, 100))
    rows = (
        (
            await db.execute(
                select(PipelineRun)
                .where(PipelineRun.pipeline_id == pipeline_id)
                .order_by(PipelineRun.id.desc())
                .limit(limit)
            )
        )
        .scalars()
        .all()
    )
    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)
async def get_run(
    run_id: int,
    _: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_READ),
    db: AsyncSession = Depends(get_db),
):
    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")
    return await _materialise_run(db, run)


@pipeline_run_router.post("/{run_id}/cancel", response_model=PipelineRunResponse)
async def cancel_run(
    run_id: int,
    _: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_RUN),
    db: AsyncSession = Depends(get_db),
):
    """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", "partial_failure"):
        return await _materialise_run(db, run)

    run.status = "cancelled"
    run.completed_at = datetime.now(timezone.utc)
    if not run.error_message:
        run.error_message = "Cancelled by user"

    job_rows = (await db.execute(select(PipelineJob).where(PipelineJob.pipeline_run_id == run.id))).scalars().all()
    for job in job_rows:
        if job.queue_entry_id:
            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"):
                queue_entry.status = "cancelled"
        if job.status not in ("completed", "failed", "cancelled"):
            job.status = "cancelled"
            job.completed_at = datetime.now(timezone.utc)

    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