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

"""API routes for File Manager (Library) functionality."""

import asyncio
import base64
import binascii
import contextlib
import hashlib
import json
import logging
import os
import re
import shutil
import uuid
import zipfile
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from pathlib import Path

from fastapi import APIRouter, Depends, File, HTTPException, Query, Response, UploadFile
from fastapi.responses import FileResponse as FastAPIFileResponse
from sqlalchemy import delete, distinct, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from backend.app.api.routes.cloud import resolve_api_key_cloud_owner
from backend.app.api.routes.library_variants import normalize_model_name, resolve_variant_model
from backend.app.api.routes.print_queue import (
    _extract_filament_types_from_3mf,
    printed_plate_of_multi_plate_file,
    unsliced_file_message,
)
from backend.app.core.auth import (
    ApiKeyActor,
    QueueReviewRequired,
    RequestActor,
    RequestPrinterScope,
    require_actor_still_exists,
    require_media_token_ownership,
    require_ownership_permission,
    require_permission_if_auth_enabled,
)
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.printer_scope import PrinterScope, ensure_model_target_allowed
from backend.app.core.row_visibility import visible_row_clauses
from backend.app.core.tasks import spawn_background_task
from backend.app.models.archive import PrintArchive
from backend.app.models.library import LibraryFile, LibraryFileTag, LibraryFolder
from backend.app.models.print_queue import PrintQueueItem
from backend.app.models.printer import Printer
from backend.app.models.project import Project
from backend.app.models.user import User
from backend.app.schemas.library import (
    AddToQueueError,
    AddToQueueRequest,
    AddToQueueResponse,
    AddToQueueResult,
    BatchThumbnailRequest,
    BatchThumbnailResponse,
    BatchThumbnailResult,
    BulkDeleteRequest,
    BulkDeleteResponse,
    ClientThumbnailResponse,
    CombineFilesRequest,
    ExternalFolderCreate,
    FileDuplicate,
    FileListResponse,
    FileMoveRequest,
    FileResponse as FileResponseSchema,
    FileUpdate,
    FileUploadResponse,
    FolderCreate,
    FolderReadmeResponse,
    FolderResponse,
    FolderTreeItem,
    FolderUpdate,
    TagSummary,
    ZipExtractError,
    ZipExtractResponse,
    ZipExtractResult,
)
from backend.app.schemas.slicer import ORCA_ONLY_BED_TYPES, SliceRequest, SliceResponse
from backend.app.services.archive import ThreeMFParser, estimate_filament_cost
from backend.app.services.design_settings import (
    DesignOverride,
    apply_design_overrides,
    extract_design_process_overrides,
    overrides_from_config,
)
from backend.app.services.filament_requirements import annotate_rack_groups
from backend.app.services.library_folder_access import (
    FolderIndex,
    can_rename_folder,
    can_write_folder,
    folder_delete_blocker,
    get_visible_folder,
    get_writable_folder,
    load_folder_index,
    sees_all_folders,
    visible_folder_ids,
)
from backend.app.services.pdf_thumbnail import generate_pdf_thumbnail
from backend.app.services.plate_thumbnail import inject_plate_thumbnails_if_missing
from backend.app.services.print_confirmation import confirm_outcome_for_new_queue_item
from backend.app.services.print_defaults import default_print_options
from backend.app.services.process_overrides import apply_process_overrides
from backend.app.services.queue_announcements import announce_queue_jobs_added
from backend.app.services.slice_output_check import (
    missing_start_gcode_message,
    start_gcode_is_missing,
    unresolved_filament_message,
    unresolved_filament_slots,
)
from backend.app.services.stl_thumbnail import MIN_USABLE_STL_BYTES, generate_stl_thumbnail
from backend.app.utils.filename import (
    MAX_FILENAME_BYTES,
    InvalidFilenameError,
    safe_path_component,
    validate_print_filename,
)
from backend.app.utils.library_paths import library_photos_dir, remove_library_photos_dir
from backend.app.utils.printer_models import is_gcode_compatible
from backend.app.utils.safe_path import PathTraversalError, assert_under, safe_join_under
from backend.app.utils.slicer_strings import first_slicer_string
from backend.app.utils.threemf_tools import (
    carries_gcode,
    default_plate_gcode_name,
    expand_to_project_slots,
    extract_embedded_presets_from_3mf,
    extract_nozzle_mapping_from_3mf,
    extract_project_filaments_from_3mf,
    is_unsliced_3mf,
    sanitize_project_settings_sentinels,
    select_plate_gcode_name,
    supports_enabled_in_config,
)

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/library", tags=["library"])

# Path of the embedded slicer config inside a BambuStudio/OrcaSlicer 3MF.
_PROJECT_SETTINGS_PATH = "Metadata/project_settings.config"


def _ensure_library_file_visible(
    library_file: LibraryFile | None,
    user: User | ApiKeyActor | None,
    can_read_all: bool,
) -> LibraryFile:
    """Per-file visibility gate for ownership-scoped LIBRARY reads (#1726-adjacent).

    Mirrors archives.py::_ensure_archive_visible — single enforcement point so a
    less-guarded sibling route can't accidentally leak a row. Same shape:

      - Missing / soft-deleted → 404 (not 403, to avoid id-enumeration leaks).
      - ``can_read_all`` true (LIBRARY_READ_ALL or auth disabled) → file returned.
      - ``can_read_all`` false and ``created_by_id != user.id`` → 404.
      - Ownerless files (``created_by_id is None``) require ALL — fail-closed.
    """
    if library_file is None or getattr(library_file, "deleted_at", None) is not None:
        raise HTTPException(404, "File not found")
    if can_read_all:
        return library_file
    if user is None:
        raise HTTPException(404, "File not found")
    if library_file.created_by_id is None or library_file.created_by_id != user.id:
        raise HTTPException(404, "File not found")
    return library_file


async def _linkable_archives(
    db: AsyncSession, archive_ids, printer_scope: PrinterScope, user: User | ApiKeyActor | None
) -> dict[int, str | None]:
    """Print names of the archives among ``archive_ids`` the caller sees in the archives.

    Live archives on the caller's printers, their own unless they see
    everyone's. A folder's link to any other archive reads as no link, and
    linking one answers as for a missing archive.
    """
    ids = {i for i in archive_ids if i}
    visible = visible_row_clauses(
        PrintArchive, printer_scope, user, Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN
    )
    if not ids or visible is None:
        return {}
    query = select(PrintArchive.id, PrintArchive.print_name).where(
        PrintArchive.id.in_(ids), PrintArchive.deleted_at.is_(None), *visible
    )
    return dict((await db.execute(query)).all())


def _archive_link(archive_id: int | None, archives: dict[int, str | None]) -> dict:
    if archive_id not in archives:
        return {"archive_id": None, "archive_name": None}
    return {"archive_id": archive_id, "archive_name": archives[archive_id]}


def _own_files_filter(user: User | ApiKeyActor | None) -> list:
    """Extra WHERE terms that count only ``user``'s files, for a read_own user.

    Folder file counts and activity times must not reveal files the user
    can't open (#3201).
    """
    if sees_all_folders(user):
        return []
    return [LibraryFile.created_by_id == user.id]


async def _load_index(db: AsyncSession, user: User | ApiKeyActor | None) -> FolderIndex:
    """The folder index a response for ``user`` needs (#3201).

    Who owns which file is needed only to decide what a read_own user sees
    and what a delete_own user may delete. Everyone else skips that scan of
    the files table, which the folder tree would otherwise pay on every load.
    """
    needs_files = user is not None and not (
        sees_all_folders(user) and user.has_permission(Permission.LIBRARY_DELETE_ALL.value)
    )
    return await load_folder_index(db, with_files=needs_files)


def _folder_access_fields(index: FolderIndex, folder: LibraryFolder, user: User | ApiKeyActor | None) -> dict:
    """The ownership fields of a folder response, for the user asking (#3201)."""
    return {
        "created_by_id": folder.created_by_id,
        "shared": bool(folder.shared),
        "can_write": can_write_folder(folder, user),
        "can_rename": can_rename_folder(folder, user),
        "can_delete": folder_delete_blocker(index, folder, user) is None,
    }


def get_library_dir() -> Path:
    """Get the library storage directory."""
    base_dir = Path(app_settings.archive_dir)
    library_dir = base_dir / "library"
    library_dir.mkdir(parents=True, exist_ok=True)
    return library_dir


def get_library_files_dir() -> Path:
    """Get the directory for library files."""
    files_dir = get_library_dir() / "files"
    files_dir.mkdir(parents=True, exist_ok=True)
    return files_dir


def classify_file_type(filename: str, file_path: Path | str | None = None) -> str:
    """Return the canonical ``LibraryFile.file_type`` for *filename*.

    Compound extensions are preserved — a `.gcode.3mf` file (a sliced
    output, still a 3MF zip on disk) is classified ``gcode.3mf`` rather
    than ``3mf``. Pre-#1600 this was only done in the external-scan
    path; the upload / ZIP-extract / in-process paths all stripped to
    the trailing extension and stored ``3mf``, so the FE had to accept
    both. Unified here so every ingest path stores the same value and
    downstream gates (gcode download, file-type filter, thumbnail
    extraction) only need to handle one canonical name per file family.
    Files with no extension classify as ``unknown``.

    Pass ``file_path`` and a ``.3mf`` is judged on what the zip actually holds
    rather than on its name (#2993). The name is not evidence: a plate exported
    from Studio or a print dispatched through the cloud reaches the archive as
    ``Foo.3mf``, G-code and all, and downloading that and re-importing it used
    to land a fully printable file in the library as a source-only project.
    A ``.gcode.3mf`` is checked as well, and one holding no G-code is a ``3mf``
    (#3435). Only the zip's file list is read, so it stays cheap.
    """
    lower = filename.lower()
    if lower.endswith(".gcode.3mf"):
        # The other way round as well (#3435): Studio saves an unsliced
        # project under whatever name it is given, and one called
        # ``Foo.gcode.3mf`` was offered Print, then refused at the queue.
        if file_path is not None and is_unsliced_3mf(file_path):
            return "3mf"
        return "gcode.3mf"
    ext = os.path.splitext(lower)[1]
    file_type = ext[1:] if ext else "unknown"
    if file_type == "3mf" and file_path is not None and carries_gcode(file_path):
        return "gcode.3mf"
    return file_type


def get_library_thumbnails_dir() -> Path:
    """Get the directory for library thumbnails."""
    thumbnails_dir = get_library_dir() / "thumbnails"
    thumbnails_dir.mkdir(parents=True, exist_ok=True)
    return thumbnails_dir


def to_relative_path(absolute_path: Path | str) -> str:
    """Convert an absolute path to a path relative to base_dir for storage."""
    if not absolute_path:
        return ""
    abs_path = Path(absolute_path)
    base_dir = Path(app_settings.base_dir)
    try:
        return str(abs_path.relative_to(base_dir))
    except ValueError:
        # Path is not under base_dir, return as-is (shouldn't happen normally)
        return str(abs_path)


def to_absolute_path(relative_path: str | None) -> Path | None:
    """Convert a relative path (from database) to an absolute path for file operations."""
    if not relative_path:
        return None
    path = Path(relative_path)
    # Handle already-absolute paths verbatim (backwards compatibility during migration).
    # Legacy DB rows may store absolute paths that predate the base_dir layout; the
    # traversal guard below only applies to relative paths coming from user input.
    if path.is_absolute():
        return path.resolve()
    base = Path(app_settings.base_dir).resolve()
    resolved = (base / relative_path).resolve()
    # Guard against path traversal — resolved path must stay inside base_dir.
    # Use is_relative_to() to avoid the /data/app vs /data/app_evil prefix confusion
    # that a plain startswith(str(base)) check would miss.
    if not resolved.is_relative_to(base):
        raise ValueError(f"Path escapes base directory: {relative_path!r}")
    return resolved


def calculate_file_hash(file_path: Path) -> str:
    """Calculate SHA256 hash of a file."""
    sha256_hash = hashlib.sha256()
    with open(file_path, "rb") as f:
        for byte_block in iter(lambda: f.read(4096), b""):
            sha256_hash.update(byte_block)
    return sha256_hash.hexdigest()


def validate_print_file_upload(filename: str, content: bytes) -> None:
    """Reject obviously-unprintable uploads early so the printer doesn't see them (#1401).

    Bambu printers in network mode only parse ``.gcode.3mf`` zip containers
    — raw ``.gcode`` and corrupt/non-zip ``.3mf`` uploads cascade into a
    confusing "Printing stopped because the printer was unable to parse the
    3mf file" rejection 30 seconds after the user clicks Print. The
    the queue dispatch path appends ``.3mf`` to a raw-gcode filename when
    constructing the FTP destination, which is how the printer ends up with a
    file named ``.gcode.3mf`` whose body is raw gcode — exactly the shape that
    triggers the firmware parse failure. Catching both classes here gives an
    actionable error at the
    upload itself.

    Compares the filename suffix rather than ``os.path.splitext`` because
    compound extensions like ``.gcode.3mf`` show up as just ``.3mf`` after
    ``splitext`` — same content validation needs to fire for both
    single-``.3mf`` and ``.gcode.3mf`` uploads.

    Raises ``HTTPException(400, ...)`` with a human-readable message on
    rejection; returns ``None`` for valid (or irrelevant — e.g. STL,
    image) uploads.
    """
    lower_filename = filename.lower()
    is_3mf_upload = lower_filename.endswith(".3mf")
    is_raw_gcode_upload = lower_filename.endswith(".gcode") and not lower_filename.endswith(".gcode.3mf")

    if is_raw_gcode_upload:
        raise HTTPException(
            status_code=400,
            detail=(
                "Raw .gcode files can't be printed on Bambu printers in network mode — "
                "they need a .gcode.3mf zip container (gcode plus metadata). Re-export from "
                "your slicer and make sure the file ends in '.gcode.3mf', not just '.gcode'. "
                "If your OS hides extensions, double-check the file with the extension visible."
            ),
        )

    if is_3mf_upload and not content.startswith(b"PK\x03\x04"):
        raise HTTPException(
            status_code=400,
            detail=(
                "This .3mf file isn't a valid ZIP container. 3MF files are ZIP archives — "
                "either the file is corrupted or it's raw gcode renamed to .3mf. Re-export "
                "from your slicer using its 'Export Plate Sliced File' action."
            ),
        )


def _resolve_upload_destination(target_folder: LibraryFolder | None, filename: str) -> tuple[Path, bool]:
    """Resolve the on-disk destination for an uploaded file.

    Non-external target: returns ``(<library_files_dir>/<uuid><ext>, False)``.
    Writable external target: writes to ``<external_path>/<filename>``
    (preserves the real filename so the file is recognisable on the mount);
    returns ``(dest, True)``. Raises ``HTTPException`` for read-only external
    folders (403), missing/inaccessible/non-writable external paths (400), and
    filename collisions on the external mount (409). See #1112 — previously
    uploads to writable external folders were silently misrouted to the
    internal library dir.
    """
    if target_folder is not None and target_folder.is_external:
        if target_folder.external_readonly:
            raise HTTPException(status_code=403, detail="Cannot upload to a read-only external folder")
        if not target_folder.external_path:
            raise HTTPException(status_code=400, detail="External folder has no configured path")
        ext_dir = Path(target_folder.external_path)
        if not ext_dir.exists() or not ext_dir.is_dir():
            raise HTTPException(
                status_code=400,
                detail=f"External path is not accessible: {target_folder.external_path}",
            )
        if not os.access(ext_dir, os.W_OK):
            raise HTTPException(
                status_code=400,
                detail=f"External path is not writable: {target_folder.external_path}",
            )
        # Guard against path-traversal via a pathological filename — join then
        # verify the resolved destination is still inside the external dir.
        dest = (ext_dir / filename).resolve()  # SEC-PATH-OK: resolve + relative_to containment check on next line
        try:
            dest.relative_to(ext_dir.resolve())
        except ValueError:
            raise HTTPException(status_code=400, detail="Invalid filename")
        if dest.exists():
            raise HTTPException(
                status_code=409,
                detail=f"A file named {filename!r} already exists in the external folder",
            )
        return dest, True
    ext = os.path.splitext(filename)[1].lower()
    return get_library_files_dir() / f"{uuid.uuid4().hex}{ext}", False


def _unique_external_name(ext_dir: Path, filename: str) -> str:
    """Return ``filename``, or the first free ``<stem> (n)<suffix>`` variant.

    Splits on the *compound* extension so re-slicing ``Bidoof.3mf`` yields
    ``Bidoof (2).gcode.3mf`` rather than ``Bidoof.gcode (2).3mf``.

    Uploads answer a name collision with a 409, which is right for a file the
    user just chose to send. A slice is not that: re-slicing the same source
    with different settings is routine, and the second run has already spent
    minutes of CPU by the time the name is known -- refusing to store it would
    throw that away. Overwriting is worse still, since the target is somebody's
    NAS and the file being replaced may not even be ours.
    """
    stem = filename[: -len(".gcode.3mf")] if filename.endswith(".gcode.3mf") else Path(filename).stem
    suffix = ".gcode.3mf" if filename.endswith(".gcode.3mf") else Path(filename).suffix
    candidate = filename
    counter = 2
    # Bounded: a directory holding 999 re-slices of one model is pathological,
    # and an unbounded loop here would hang the request on a mount that lies
    # about exists() (some SMB shares do under contention).
    #
    # safe_join_under rather than `ext_dir / candidate`: `filename` derives
    # from a name read out of a 3MF, so the very first probe must not be able
    # to stat its way outside the mount. It raises PathTraversalError, which
    # the caller turns into a managed-storage fallback.
    while safe_join_under(ext_dir, candidate, http=False).exists() and counter < 1000:
        candidate = f"{stem} ({counter}){suffix}"
        counter += 1
    return candidate


def _resolve_slice_destination(target_folder: LibraryFolder | None, out_filename: str) -> tuple[Path, bool, str | None]:
    """Resolve where a slice result should be written.

    Returns ``(path, is_external, fallback_reason)``. ``fallback_reason`` is
    ``None`` on the normal paths and otherwise names why an external folder
    could not receive the file, so the caller can tell the user instead of
    quietly filing it elsewhere.

    Slicing a file that lives on an external mount used to store the output in
    the managed library dir unconditionally, while giving the new row the
    external folder's ``folder_id`` (#2810). The file therefore appeared in the
    right folder in the UI and never arrived on the share, which is the one
    place the user was looking -- and made it un-reproducible from the web UI
    alone. Uploads learned this in #1112 (``_resolve_upload_destination``) and
    moves in its follow-up (``_move_file_bytes``); slicing was the last write
    path still assuming managed storage.

    Unlike uploads, a failure here does not raise. The bytes exist and cost
    real time to produce, so an unwritable target falls back to managed storage
    with a reason attached rather than discarding the slice.
    """
    if target_folder is None or not target_folder.is_external:
        return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, None

    if target_folder.external_readonly:
        return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, "external_readonly"
    if not target_folder.external_path:
        return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, "external_no_path"

    ext_dir = Path(target_folder.external_path)
    if not ext_dir.exists() or not ext_dir.is_dir():
        return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, "external_unreachable"
    if not os.access(ext_dir, os.W_OK):
        return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, "external_not_writable"

    try:
        dest = safe_join_under(ext_dir, _unique_external_name(ext_dir, out_filename), http=False)
    except PathTraversalError:
        # The source filename reached us from a 3MF on disk, so this is
        # defensive rather than expected -- but a name that escapes the mount
        # must land in managed storage, never outside it.
        return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, "external_invalid_name"
    return dest, True, None


async def _folder_tree_file_ids(db: AsyncSession, folder_id: int) -> list[int]:
    """Every ``LibraryFile`` id under ``folder_id``, at any depth.

    Deleting a folder cascades to its whole subtree, so anything that has to be
    released before that delete (queue items, cross-model candidates) needs the
    subtree, not just the folder's own files.

    Trashed rows are included deliberately: they are still real rows and the
    cascade takes them too.
    """
    file_ids: list[int] = []
    pending = [folder_id]
    # The API refuses to make a folder its own ancestor, so a loop here would
    # mean the table is already corrupt -- but this walk runs inside a delete
    # request, and hanging one is worse than the cost of a set.
    seen: set[int] = set()
    while pending:
        current = pending.pop()
        if current in seen:
            continue
        seen.add(current)
        file_ids.extend(
            (await db.execute(select(LibraryFile.id).where(LibraryFile.folder_id == current))).scalars().all()
        )
        pending.extend(
            (await db.execute(select(LibraryFolder.id).where(LibraryFolder.parent_id == current))).scalars().all()
        )
    return file_ids


def _stored_file_path(abs_path: Path, is_external: bool) -> str:
    """Produce the value to persist in ``LibraryFile.file_path``.

    External files store the absolute mount path directly (same as scan does),
    so ``to_absolute_path`` round-trips through its ``is_absolute()`` fast
    path. Managed files store a path relative to ``base_dir`` for portability.
    """
    return str(abs_path) if is_external else to_relative_path(abs_path)


class _MoveSkip(Exception):
    """Signalled by ``_move_file_bytes`` to skip a file with a user-visible reason.

    Carries an optional `code` for machine-friendly grouping (the
    front-end can localise it) and a fallback English `reason` for logs.
    """

    def __init__(self, code: str, reason: str):
        super().__init__(reason)
        self.code = code
        self.reason = reason


def _resolve_source_disk_path(file: LibraryFile) -> Path | None:
    """Return the absolute on-disk path for an existing LibraryFile, or None
    if it can't be located (legacy DB row, deleted file, etc.)."""
    if file.is_external:
        return Path(file.file_path) if file.file_path else None
    return to_absolute_path(file.file_path)


def _move_file_bytes(file: LibraryFile, target_folder: LibraryFolder | None) -> str:
    """Physically relocate `file`'s bytes to match `target_folder`.

    Used by the move endpoint when source/target straddle the
    managed↔external boundary (#1112 follow-up — the prior implementation
    updated the DB row's ``folder_id`` but never moved the bytes, so a
    file moved to an external SMB folder showed up in Bambuddy's UI but
    not on the NAS).

    Returns the new ``file_path`` value to persist (relative for managed
    targets, absolute for external targets — matches the upload + scan
    paths). Raises ``_MoveSkip`` for any condition that would make the
    move unsafe (target unwritable, filename collision, source missing).

    The copy-then-unlink ordering means a partial copy followed by a
    failed unlink leaves both the source and the dest on disk — better
    than the symmetric "rename or move" which would lose the source if
    the target write didn't complete on a flaky mount. The DB row stays
    pointed at the source until the caller commits the new ``file_path``.
    """
    src = _resolve_source_disk_path(file)
    if not src or not src.exists():
        raise _MoveSkip("source_missing", "source file missing on disk")

    target_is_external = target_folder is not None and target_folder.is_external

    if target_is_external:
        if target_folder.external_readonly:
            # Already blocked at top level, but defence-in-depth.
            raise _MoveSkip("target_readonly", "target external folder is read-only")
        if not target_folder.external_path:
            raise _MoveSkip("target_misconfigured", "target external folder has no path")
        ext_dir = Path(target_folder.external_path)
        if not ext_dir.exists() or not ext_dir.is_dir():
            raise _MoveSkip("target_inaccessible", f"target path not accessible: {ext_dir}")
        if not os.access(ext_dir, os.W_OK):
            raise _MoveSkip("target_unwritable", f"target path not writable: {ext_dir}")
        dest = (ext_dir / file.filename).resolve()  # SEC-PATH-OK: resolve + relative_to containment check on next line
        try:
            dest.relative_to(ext_dir.resolve())
        except ValueError:
            raise _MoveSkip("invalid_filename", f"unsafe filename: {file.filename!r}") from None
        if dest.exists():
            raise _MoveSkip("name_collision", f"a file named {file.filename!r} already exists in target")
        try:
            shutil.copy2(src, dest)
        except OSError as e:
            # Clean up partial dest so a retry can succeed.
            with contextlib.suppress(OSError):
                dest.unlink(missing_ok=True)
            raise _MoveSkip("copy_failed", f"copy failed: {e}") from e
    else:
        # → managed (root or non-external folder): generate a fresh UUID
        # filename in the internal store so we don't collide with another
        # file that happens to share `filename`.
        ext = src.suffix.lower()
        dest = get_library_files_dir() / f"{uuid.uuid4().hex}{ext}"
        try:
            shutil.copy2(src, dest)
        except OSError as e:
            with contextlib.suppress(OSError):
                dest.unlink(missing_ok=True)
            raise _MoveSkip("copy_failed", f"copy failed: {e}") from e

    # Copy succeeded — unlink the original. A failure here leaves an
    # orphan on disk but the DB row is consistent against the new dest.
    try:
        src.unlink(missing_ok=True)
    except OSError as e:
        logger.warning(
            "Move: copied %s → %s but couldn't remove source: %s",
            src,
            dest,
            e,
        )

    return _stored_file_path(dest, is_external=target_is_external)


def _clean_3mf_metadata(obj):
    """Strip bytes and thumbnail-carrier keys so the payload is JSON-storable.

    Shared by ``upload_file`` and :func:`save_3mf_bytes_to_library` — the
    ``ThreeMFParser`` output embeds the thumbnail bytes under
    ``_thumbnail_data``/``_thumbnail_ext`` and may also include raw bytes in
    other fields, none of which can be JSON-encoded.
    """
    if isinstance(obj, dict):
        return {
            k: _clean_3mf_metadata(v)
            for k, v in obj.items()
            if not isinstance(v, bytes) and k not in ("_thumbnail_data", "_thumbnail_ext")
        }
    if isinstance(obj, list):
        return [_clean_3mf_metadata(i) for i in obj if not isinstance(i, bytes)]
    if isinstance(obj, bytes):
        return None
    return obj


def _read_3mf_entry(zip_path: Path, entry: str) -> bytes | None:
    """Return the raw bytes of an entry inside a 3MF (ZIP), or ``None`` when
    the file isn't a parseable zip / doesn't contain that entry / any IO
    error. Used to lift the source archive's per-plate render onto a
    re-sliced archive (#1493 follow-up) — the slicer CLI often doesn't
    emit a fresh ``Metadata/plate_N.png`` and the project-wide cover-art
    fallback in :class:`ThreeMFParser` looks unrelated to the actual slice.
    """
    try:
        with zipfile.ZipFile(zip_path, "r") as zf:
            if entry not in zf.namelist():
                return None
            return zf.read(entry)
    except (zipfile.BadZipFile, OSError, KeyError):
        return None


def _without_print_name(metadata: dict | None) -> dict | None:
    """Drop the embedded 3MF Title (``print_name``) from library-file metadata.

    The 3MF ``<metadata name="Title">`` holds the in-app project title — the
    generic ``"Exported 3D Model"`` for a Bambu Studio "Save As", a marketing
    title for a MakerWorld download — never the filename the user saved as.
    The FileManager keys its display name, search and sort off ``print_name``,
    so storing it makes every card show the wrong name (#1489). A library
    file's display name is its filename; only ``PrintArchive`` carries a real
    ``print_name``. Returns the input unchanged when there's nothing to strip;
    otherwise a new dict (never mutates the argument).
    """
    if not metadata or "print_name" not in metadata:
        return metadata
    return {k: v for k, v in metadata.items() if k != "print_name"}


async def save_3mf_bytes_to_library(
    db: AsyncSession,
    *,
    file_bytes: bytes | None = None,
    source_path: Path | None = None,
    filename: str,
    folder_id: int | None = None,
    source_type: str | None = None,
    source_url: str | None = None,
    owner_id: int | None = None,
    dedupe_within: list | tuple | None = (),
) -> tuple[LibraryFile, bool]:
    """Save a fetched file into the library and return ``(library_file, was_existing)``.

    Used by routes that receive a file in-process rather than as a multipart
    upload: the MakerWorld import (3MF) and the Manyfold import (3MF, STL,
    STEP). Deduplicates by ``source_url`` when provided —
    if a LibraryFile with the same source_url already exists, the existing
    row is returned and the bytes are NOT re-saved (MakerWorld signed URLs
    change each download, so hash-based dedupe alone would miss re-imports).
    ``dedupe_within`` narrows the rows that count as that copy to the ones the
    importer may open; None imports a copy of their own regardless.

    Parses 3MF metadata + thumbnail the same way the multipart upload route
    does, via :class:`ThreeMFParser`, and renders an STL's thumbnail. Paths are stored as relative so the
    library is portable across installs.

    Give the content as ``file_bytes`` or as ``source_path``, a file on disk
    that is copied without being read into memory (a virtual printer's
    upload, which can be hundreds of MB).
    """
    if (file_bytes is None) == (source_path is None):
        raise ValueError("pass exactly one of file_bytes and source_path")
    # Source-URL-based dedupe: return the existing row untouched.
    if source_url and dedupe_within is not None:
        existing = await db.execute(
            LibraryFile.active().where(LibraryFile.source_url == source_url, *dedupe_within).limit(1)
        )
        existing_row = existing.scalar_one_or_none()
        if existing_row is not None:
            return existing_row, True

    # Resolve target folder so writable-external destinations land on the
    # mount with the real filename, instead of being silently misrouted to
    # the internal library dir with a UUID name (#1645). Mirrors what the
    # multipart-upload path has done since #1112. ``_resolve_upload_destination``
    # also enforces the 403 read-only / 400 unwritable / 409 collision
    # rejections — the makerworld route layer already pre-checks read-only,
    # but the helper's checks remain as defence-in-depth for any future
    # caller that skips that route gate.
    target_folder: LibraryFolder | None = None
    if folder_id is not None:
        folder_q = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
        target_folder = folder_q.scalar_one_or_none()

    # Another user's copy of this very import, which the importer may not
    # open, may already sit on the mount under this name: theirs is stored
    # beside it under a free name, as a slice is. Any other file of that name
    # is still refused, so nothing put there outside Bambuddy is touched.
    if (
        source_url
        and target_folder is not None
        and target_folder.is_external
        and target_folder.external_path
        and Path(target_folder.external_path).is_dir()
        and await db.scalar(
            select(LibraryFile.id)
            .where(LibraryFile.source_url == source_url, LibraryFile.deleted_at.is_(None))
            .limit(1)
        )
        is not None
    ):
        with contextlib.suppress(PathTraversalError):
            filename = _unique_external_name(Path(target_folder.external_path), filename)

    file_path, is_external = _resolve_upload_destination(target_folder, filename)
    ext = file_path.suffix.lower() or ".3mf"
    if source_path is not None:
        await asyncio.to_thread(shutil.copyfile, source_path, file_path)
        file_size = file_path.stat().st_size
    else:
        assert file_bytes is not None  # one of the two, checked above
        with open(file_path, "wb") as fh:
            fh.write(file_bytes)
        file_size = len(file_bytes)

    file_hash = calculate_file_hash(file_path)

    # Extract metadata + thumbnail from the 3MF.
    metadata: dict | None = None
    thumbnail_path: str | None = None
    if ext == ".3mf":
        try:
            # Off the event loop: the parser still decompresses the whole model
            # entry, which for a combined plate is hundreds of MB (#3162).
            parser = ThreeMFParser(str(file_path))
            raw_metadata = await asyncio.to_thread(parser.parse)
            thumb_data = raw_metadata.get("_thumbnail_data")
            thumb_ext = raw_metadata.get("_thumbnail_ext", ".png")
            if thumb_data:
                thumbs_dir = get_library_thumbnails_dir()
                thumb_filename = f"{uuid.uuid4().hex}{thumb_ext}"
                thumb_path = thumbs_dir / thumb_filename  # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + thumb_ext
                with open(thumb_path, "wb") as fh:
                    fh.write(thumb_data)
                thumbnail_path = str(thumb_path)
            metadata = _clean_3mf_metadata(raw_metadata) or None
        except Exception as exc:
            # Matches the multipart upload route's behaviour — a bad 3MF should
            # still land in the library so the user can see / delete it rather
            # than failing the whole request.
            logger.warning("Failed to parse 3MF %s: %s", filename, exc)
    elif ext == ".stl":
        # Manyfold imports (#1471) bring STLs. Same thumbnail as the multipart
        # upload gives them, with the same pre-skip for stubs too small to hold
        # a triangle.
        try:
            if file_path.stat().st_size >= MIN_USABLE_STL_BYTES:
                thumbnail_path = await _render_stl_thumbnail(file_path, get_library_thumbnails_dir())
        except Exception as exc:  # noqa: BLE001 — a thumbnail must never fail the import
            logger.warning("Failed to render STL thumbnail for %s: %s", filename, exc)

    library_file = LibraryFile(
        folder_id=folder_id,
        is_external=is_external,
        filename=filename,
        file_path=_stored_file_path(file_path, is_external),
        file_type=classify_file_type(filename, file_path),
        file_size=file_size,
        file_hash=file_hash,
        thumbnail_path=to_relative_path(thumbnail_path) if thumbnail_path else None,
        file_metadata=_without_print_name(metadata),
        source_type=source_type,
        source_url=source_url,
        created_by_id=owner_id,
    )
    db.add(library_file)
    await db.commit()
    await db.refresh(library_file)
    return library_file, False


def extract_gcode_thumbnail(file_path: Path) -> bytes | None:
    """Extract embedded thumbnail from gcode file.

    Supports PrusaSlicer/BambuStudio format:
    ; thumbnail begin WxH SIZE
    ; base64data...
    ; thumbnail end
    """
    try:
        thumbnail_data = None
        in_thumbnail = False
        thumbnail_lines = []
        best_size = 0

        with open(file_path, errors="ignore") as f:
            # Only read first 50KB for performance (thumbnails are at the start)
            content = f.read(50000)

        for line in content.split("\n"):
            line = line.strip()

            # Check for thumbnail start
            if line.startswith("; thumbnail begin"):
                in_thumbnail = True
                thumbnail_lines = []
                # Parse dimensions: "; thumbnail begin 300x300 12345"
                match = re.search(r"(\d+)x(\d+)", line)
                if match:
                    width = int(match.group(1))
                    # Prefer larger thumbnails (up to 300px)
                    if width > best_size and width <= 300:
                        best_size = width
                continue

            # Check for thumbnail end
            if line.startswith("; thumbnail end"):
                if in_thumbnail and thumbnail_lines:
                    try:
                        # Decode the base64 data
                        b64_data = "".join(thumbnail_lines)
                        decoded = base64.b64decode(b64_data)
                        # Only keep if this is the best size or first valid thumbnail
                        if thumbnail_data is None or best_size > 0:
                            thumbnail_data = decoded
                    except (binascii.Error, ValueError):
                        pass  # Skip thumbnail with invalid base64 data
                in_thumbnail = False
                thumbnail_lines = []
                continue

            # Collect thumbnail data
            if in_thumbnail and line.startswith(";"):
                # Remove the leading "; " or ";"
                data_line = line[1:].strip()
                if data_line:
                    thumbnail_lines.append(data_line)

        return thumbnail_data
    except Exception as e:
        logger.warning("Failed to extract gcode thumbnail: %s", e)
        return None


def create_image_thumbnail(file_path: Path, thumbnails_dir: Path, max_size: int = 256) -> str | None:
    """Create a thumbnail from an image file.

    For small images, copies directly. For larger images, resizes.
    Returns the thumbnail path or None on failure.
    """
    try:
        from PIL import Image

        thumb_filename = f"{uuid.uuid4().hex}.png"
        thumb_path = thumbnails_dir / thumb_filename  # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + ".png"

        with Image.open(file_path) as img:
            # Convert to RGB if necessary (for PNG with transparency, etc.)
            if img.mode in ("RGBA", "LA", "P"):
                # Create white background for transparency
                background = Image.new("RGB", img.size, (255, 255, 255))
                if img.mode == "P":
                    img = img.convert("RGBA")
                background.paste(img, mask=img.split()[-1] if img.mode == "RGBA" else None)
                img = background
            elif img.mode != "RGB":
                img = img.convert("RGB")

            # Resize if larger than max_size
            if img.width > max_size or img.height > max_size:
                img.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)

            img.save(thumb_path, "PNG", optimize=True)

        return str(thumb_path)
    except ImportError:
        # PIL not installed, just copy the file if it's small enough
        logger.warning("PIL not installed, copying image as thumbnail")
        try:
            file_size = file_path.stat().st_size
            if file_size < 500000:  # Less than 500KB
                thumb_filename = f"{uuid.uuid4().hex}{file_path.suffix}"
                thumb_path = (
                    thumbnails_dir / thumb_filename
                )  # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + file_path.suffix
                shutil.copy2(file_path, thumb_path)
                return str(thumb_path)
        except OSError:
            pass  # File inaccessible; fall through to return None
        return None
    except Exception as e:
        logger.warning("Failed to create image thumbnail: %s", e)
        return None


# Supported image extensions for thumbnails
IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".tif"}

# File types whose thumbnails are rendered client-side and uploaded back
# (#2976). The server has no renderer for STEP (that would need OpenCascade)
# or the spreadsheet types, so the browser posts its first preview render to
# POST /files/{id}/preview-thumbnail instead. PDF is rendered server-side with
# PDFium when it lands and stays here for a PDF that renderer cannot read.
# Kept to exactly these types so the endpoint can never overwrite a
# server-generated STL/3MF/G-code/image thumbnail.
CLIENT_THUMBNAIL_TYPES = {"step", "stp", "pdf", "csv", "xlsx", "ods"}

# Photos of the printed result (#3077): same allowlist and naming as the
# archive photo routes. 10 MB is ample for a phone camera JPEG.
PHOTO_EXTENSIONS = (".jpg", ".jpeg", ".png", ".webp")
PHOTO_MEDIA_TYPES = {
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".png": "image/png",
    ".webp": "image/webp",
}
MAX_PHOTO_BYTES = 10 * 1024 * 1024

# Upper bound for an uploaded client-rendered thumbnail. The FE sends a
# 256px PNG (a few tens of KB); anything near this limit is not a thumbnail.
MAX_CLIENT_THUMBNAIL_BYTES = 2 * 1024 * 1024

# Upper bound on the *decoded* size, checked against the header before any
# pixels are allocated: a few-KB PNG can declare 12000x7000 and still be under
# PIL's own decompression-bomb limit, which would be ~340 MB of RGBA.
MAX_CLIENT_THUMBNAIL_EDGE = 2048

# What the endpoint stores. The grid renders at ~256px, so anything larger is
# downscaled rather than kept.
STORED_CLIENT_THUMBNAIL_EDGE = 512


# STL thumbnails get their own single worker thread. They are made one at a
# time anyway (stl_thumbnail._RENDER_LOCK), and in the shared default pool the
# ones waiting their turn would hold threads that every other to_thread call in
# the app needs: a burst of STL uploads would stall them all.
_STL_THUMBNAIL_EXECUTOR = ThreadPoolExecutor(max_workers=1, thread_name_prefix="stl-thumbnail")


async def _render_stl_thumbnail(stl_path: Path, thumbnails_dir: Path) -> str | None:
    """generate_stl_thumbnail, off the event loop: a large mesh takes seconds."""
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(_STL_THUMBNAIL_EXECUTOR, generate_stl_thumbnail, stl_path, thumbnails_dir)


async def _backfill_external_thumbnails(folder_ids: list[int]) -> None:
    """Generate STL and PDF thumbnails for an external folder tree in the background.

    Spawned via ``asyncio.create_task`` from ``scan_external_folder`` so the
    HTTP request can return as soon as the filesystem walk + folder/file rows
    are committed. Thumbnails for thousands of STL files would otherwise hold
    the request open for many minutes (each file triggers a ``trimesh.load``
    + matplotlib render, ~1-5s each) and the FE modal times out before the
    final ``db.commit()`` runs — causing the original symptom in #1299 where
    subdirectories never showed up because nothing got committed. PDFs are
    faster (a PDFium page render) but a share holding hundreds of them would
    still hold the request open, so they are rendered here too.

    Opens its own session because the request session is closed by the time
    this task starts running. Commits per-file so a worker restart mid-run
    only loses the in-flight file. Caps STL load to a single file at a time
    to avoid memory pressure on systems with many huge STLs.
    """
    if not folder_ids:
        return
    thumbnails_dir = get_library_thumbnails_dir()
    async with async_session() as db:
        result = await db.execute(
            LibraryFile.active().where(
                LibraryFile.folder_id.in_(folder_ids),
                LibraryFile.file_type.in_(("stl", "pdf")),
                LibraryFile.thumbnail_path.is_(None),
            )
        )
        target_files = result.scalars().all()
        if not target_files:
            return
        logger.info(
            "Backfilling STL/PDF thumbnails: %d file(s) across %d folder(s)",
            len(target_files),
            len(folder_ids),
        )
        for target_file in target_files:
            abs_path = to_absolute_path(target_file.file_path)
            if not abs_path or not abs_path.exists():
                continue
            if target_file.file_type == "pdf":
                # generate_pdf_thumbnail never raises; an unreadable PDF
                # returns None and keeps the browser-preview fallback.
                thumb_path = await asyncio.to_thread(generate_pdf_thumbnail, abs_path, thumbnails_dir)
                if thumb_path:
                    target_file.thumbnail_path = to_relative_path(Path(thumb_path))
                    await db.commit()
                continue
            # Pre-skip files too small to contain even a single triangle.
            # Bulk-uploaded ZIPs of stub STLs would otherwise trigger one
            # trimesh.load() call + one debug log line per stub.
            try:
                if abs_path.stat().st_size < MIN_USABLE_STL_BYTES:
                    continue
            except OSError:
                continue
            try:
                thumb_path = await _render_stl_thumbnail(abs_path, thumbnails_dir)
            except Exception as exc:  # noqa: BLE001 — never let one bad STL kill the rest
                logger.debug("STL thumbnail backfill skipped %s: %s", abs_path, exc)
                continue
            if thumb_path:
                target_file.thumbnail_path = to_relative_path(Path(thumb_path))
                await db.commit()


# ============ Folder Endpoints ============


@router.get("/folders", response_model=list[FolderTreeItem])
@router.get("/folders/", response_model=list[FolderTreeItem])
async def list_folders(
    response: Response,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
    printer_scope: PrinterScope = RequestPrinterScope,
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Get all folders as a tree structure.

    A library:read_own user gets only the folders they may see (#3201), and
    the file counts and activity times of their own files only.
    """
    user, _ = auth_result
    # Prevent browser caching of folder list
    response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"

    index = await _load_index(db, user)
    visible = visible_folder_ids(index, user)

    # Get all folders with project and archive joins
    result = await db.execute(
        select(LibraryFolder, Project.name)
        .outerjoin(Project, LibraryFolder.project_id == Project.id)
        .order_by(LibraryFolder.name)
    )
    rows = [row for row in result.all() if visible is None or row[0].id in visible]
    archives = await _linkable_archives(db, [row[0].archive_id for row in rows], printer_scope, actor)
    own_files = _own_files_filter(user)

    # Get file counts per folder
    file_counts_result = await db.execute(
        select(LibraryFile.folder_id, func.count(LibraryFile.id))
        .where(LibraryFile.folder_id.isnot(None), LibraryFile.deleted_at.is_(None), *own_files)
        .group_by(LibraryFile.folder_id)
    )
    file_counts = dict(file_counts_result.all())

    # Latest immediate-child file activity per folder (#1770/#2680). Real on-disk
    # mtime when we have it (external scans populate ``fs_modified_at``), else the
    # DB ``updated_at`` — COALESCE so external rows scanned before this field
    # existed, and internal uploads, still contribute a signal. This is the
    # per-folder *leaf* value; subtree descent is aggregated recursively below.
    latest_file_activity_result = await db.execute(
        select(
            LibraryFile.folder_id,
            func.max(func.coalesce(LibraryFile.fs_modified_at, LibraryFile.updated_at)),
        )
        .where(LibraryFile.folder_id.isnot(None), LibraryFile.deleted_at.is_(None), *own_files)
        .group_by(LibraryFile.folder_id)
    )
    latest_file_activity = dict(latest_file_activity_result.all())

    # Build tree structure. Each folder's initial ``latest_activity_at`` is its own
    # leaf activity: the newer of its real directory mtime (fallback updated_at)
    # and its immediate files' mtime. The recursive bubble below then rolls each
    # subtree's newest descendant up to its ancestors (#2680 — sorting must match
    # ``ls -t`` recursively, so a freshly-added deep file lifts every parent).
    folder_map = {}
    root_folders = []

    for folder, project_name in rows:
        own_activity = folder.fs_modified_at or folder.updated_at
        latest_file = latest_file_activity.get(folder.id)
        if latest_file is not None and latest_file > own_activity:
            own_activity = latest_file
        folder_item = FolderTreeItem(
            id=folder.id,
            name=folder.name,
            parent_id=folder.parent_id,
            project_id=folder.project_id,
            project_name=project_name,
            **_archive_link(folder.archive_id, archives),
            is_external=folder.is_external,
            external_path=folder.external_path,
            external_readonly=folder.external_readonly,
            file_count=file_counts.get(folder.id, 0),
            latest_activity_at=own_activity,
            **_folder_access_fields(index, folder, user),
            children=[],
        )
        folder_map[folder.id] = folder_item

    # Link children to parents
    for folder, _ in rows:
        folder_item = folder_map[folder.id]
        if folder.parent_id is None:
            root_folders.append(folder_item)
        elif folder.parent_id in folder_map:
            folder_map[folder.parent_id].children.append(folder_item)

    # Recursive newest-descendant bubble (#2680). Post-order: a folder's activity
    # becomes the max of its own leaf activity and every descendant's, so sorting
    # the tree by ``latest_activity_at`` surfaces the branch with the most recent
    # activity anywhere inside it. Iterative stack keeps deep external mounts off
    # Python's recursion limit.
    def _bubble(root: FolderTreeItem) -> None:
        order: list[FolderTreeItem] = []
        stack = [root]
        while stack:
            node = stack.pop()
            order.append(node)
            stack.extend(node.children)
        for node in reversed(order):  # deepest first
            for child in node.children:
                if child.latest_activity_at is not None and (
                    node.latest_activity_at is None or child.latest_activity_at > node.latest_activity_at
                ):
                    node.latest_activity_at = child.latest_activity_at

    for root in root_folders:
        _bubble(root)

    return root_folders


@router.get("/folders/by-project/{project_id}", response_model=list[FolderResponse])
async def get_folders_by_project(
    project_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
    printer_scope: PrinterScope = RequestPrinterScope,
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Get all folders linked to a specific project, those the user may see (#3201)."""
    user, _ = auth_result
    index = await _load_index(db, user)
    visible = visible_folder_ids(index, user)
    own_files = _own_files_filter(user)
    result = await db.execute(
        select(LibraryFolder, Project.name)
        .outerjoin(Project, LibraryFolder.project_id == Project.id)
        .where(LibraryFolder.project_id == project_id)
        .order_by(LibraryFolder.name)
    )
    rows = [row for row in result.all() if visible is None or row[0].id in visible]
    archives = await _linkable_archives(db, [row[0].archive_id for row in rows], printer_scope, actor)

    folders = []
    for folder, project_name in rows:
        # Get file count + latest file activity (#1770/#2680) in one trip. Prefer
        # the real on-disk mtime (external scans), fall back to the DB updated_at.
        agg_result = await db.execute(
            select(
                func.count(LibraryFile.id),
                func.max(func.coalesce(LibraryFile.fs_modified_at, LibraryFile.updated_at)),
            ).where(
                LibraryFile.folder_id == folder.id,
                LibraryFile.deleted_at.is_(None),
                *own_files,
            )
        )
        file_count, latest_file = agg_result.one()
        file_count = file_count or 0
        own_activity = folder.fs_modified_at or folder.updated_at
        latest_activity_at = max(own_activity, latest_file) if latest_file is not None else own_activity

        folders.append(
            FolderResponse(
                id=folder.id,
                name=folder.name,
                parent_id=folder.parent_id,
                project_id=folder.project_id,
                project_name=project_name,
                **_archive_link(folder.archive_id, archives),
                is_external=folder.is_external,
                external_path=folder.external_path,
                external_readonly=folder.external_readonly,
                external_show_hidden=folder.external_show_hidden,
                file_count=file_count,
                latest_activity_at=latest_activity_at,
                **_folder_access_fields(index, folder, user),
                created_at=folder.created_at,
                updated_at=folder.updated_at,
            )
        )

    return folders


@router.get("/folders/by-archive/{archive_id}", response_model=list[FolderResponse])
async def get_folders_by_archive(
    archive_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
    printer_scope: PrinterScope = RequestPrinterScope,
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Get all folders linked to a specific archive, those the user may see (#3201)."""
    user, _ = auth_result
    return (await _folders_by_archive(db, user, [archive_id], printer_scope, actor))[archive_id]


async def _folders_by_archive(
    db: AsyncSession,
    user: User | ApiKeyActor | None,
    archive_ids: list[int],
    printer_scope: PrinterScope,
    actor: User | ApiKeyActor | None,
) -> dict[int, list]:
    """The folders linked to each archive that *user* may see (#3201), by archive id.

    An archive *actor* (the caller, as ``RequestActor`` gives it) does not see
    in the archives has none.
    """
    folders: dict[int, list] = {archive_id: [] for archive_id in archive_ids}
    archives = await _linkable_archives(db, archive_ids, printer_scope, actor)
    if not archives:
        return folders
    index = await _load_index(db, user)
    visible = visible_folder_ids(index, user)
    own_files = _own_files_filter(user)
    result = await db.execute(
        select(LibraryFolder).where(LibraryFolder.archive_id.in_(list(archives))).order_by(LibraryFolder.name)
    )
    rows = [folder for folder in result.scalars().all() if visible is None or folder.id in visible]

    for folder in rows:
        # Get file count + latest file activity (#1770/#2680) in one trip. Prefer
        # the real on-disk mtime (external scans), fall back to the DB updated_at.
        agg_result = await db.execute(
            select(
                func.count(LibraryFile.id),
                func.max(func.coalesce(LibraryFile.fs_modified_at, LibraryFile.updated_at)),
            ).where(
                LibraryFile.folder_id == folder.id,
                LibraryFile.deleted_at.is_(None),
                *own_files,
            )
        )
        file_count, latest_file = agg_result.one()
        file_count = file_count or 0
        own_activity = folder.fs_modified_at or folder.updated_at
        latest_activity_at = max(own_activity, latest_file) if latest_file is not None else own_activity

        assert folder.archive_id is not None  # selected by archive_id
        folders[folder.archive_id].append(
            FolderResponse(
                id=folder.id,
                name=folder.name,
                parent_id=folder.parent_id,
                project_id=folder.project_id,
                project_name=None,
                **_archive_link(folder.archive_id, archives),
                is_external=folder.is_external,
                external_path=folder.external_path,
                external_readonly=folder.external_readonly,
                external_show_hidden=folder.external_show_hidden,
                file_count=file_count,
                latest_activity_at=latest_activity_at,
                **_folder_access_fields(index, folder, user),
                created_at=folder.created_at,
                updated_at=folder.updated_at,
            )
        )

    return folders


@router.post("/folders", response_model=FolderResponse)
@router.post("/folders/", response_model=FolderResponse)
async def create_folder(
    data: FolderCreate,
    db: AsyncSession = Depends(get_db, scope="function"),
    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
    actor: User | ApiKeyActor | None = RequestActor,
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Create a new folder, owned by the user who makes it (#3201)."""
    # A read_own user may only create inside their own or a shared folder.
    if data.parent_id is not None:
        await get_writable_folder(db, data.parent_id, actor)

    # Verify project exists if specified
    project_name = None
    if data.project_id is not None:
        project_result = await db.execute(select(Project).where(Project.id == data.project_id))
        project = project_result.scalar_one_or_none()
        if not project:
            raise HTTPException(status_code=404, detail="Project not found")
        project_name = project.name

    # Verify archive exists if specified
    archives = await _linkable_archives(db, [data.archive_id], printer_scope, actor)
    if data.archive_id is not None and data.archive_id not in archives:
        raise HTTPException(status_code=404, detail="Archive not found")

    folder = LibraryFolder(
        name=data.name,
        parent_id=data.parent_id,
        project_id=data.project_id,
        archive_id=data.archive_id,
        created_by_id=actor.id if actor else None,
        # Made without a user (auth off, a key without an owner): everyone's, as before #3201.
        shared=actor is None or actor.id is None,
    )
    db.add(folder)
    await db.commit()
    await db.refresh(folder)
    index = await _load_index(db, actor)

    return FolderResponse(
        id=folder.id,
        name=folder.name,
        parent_id=folder.parent_id,
        project_id=folder.project_id,
        project_name=project_name,
        **_archive_link(folder.archive_id, archives),
        is_external=folder.is_external,
        external_path=folder.external_path,
        external_readonly=folder.external_readonly,
        external_show_hidden=folder.external_show_hidden,
        file_count=0,
        # New folder has no files yet — fall back to the folder's own
        # updated_at so this matches the list-route semantics (#1770).
        latest_activity_at=folder.updated_at,
        **_folder_access_fields(index, folder, actor),
        created_at=folder.created_at,
        updated_at=folder.updated_at,
    )


@router.get("/folders/{folder_id}", response_model=FolderResponse)
async def get_folder(
    folder_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
    printer_scope: PrinterScope = RequestPrinterScope,
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Get a folder by ID; 404 for a folder the user may not see (#3201)."""
    user, _ = auth_result
    index = await _load_index(db, user)
    visible = visible_folder_ids(index, user)
    if visible is not None and folder_id not in visible:
        raise HTTPException(status_code=404, detail="Folder not found")
    result = await db.execute(
        select(LibraryFolder, Project.name)
        .outerjoin(Project, LibraryFolder.project_id == Project.id)
        .where(LibraryFolder.id == folder_id)
    )
    row = result.one_or_none()

    if not row:
        raise HTTPException(status_code=404, detail="Folder not found")

    folder, project_name = row
    archives = await _linkable_archives(db, [folder.archive_id], printer_scope, actor)

    # Get file count + latest file activity (#1770) in one trip
    agg_result = await db.execute(
        select(
            func.count(LibraryFile.id),
            func.max(LibraryFile.updated_at),
        ).where(
            LibraryFile.folder_id == folder_id,
            LibraryFile.deleted_at.is_(None),
            *_own_files_filter(user),
        )
    )
    file_count, latest_file = agg_result.one()
    file_count = file_count or 0
    latest_activity_at = max(folder.updated_at, latest_file) if latest_file is not None else folder.updated_at

    return FolderResponse(
        id=folder.id,
        name=folder.name,
        parent_id=folder.parent_id,
        project_id=folder.project_id,
        project_name=project_name,
        **_archive_link(folder.archive_id, archives),
        is_external=folder.is_external,
        external_path=folder.external_path,
        external_readonly=folder.external_readonly,
        external_show_hidden=folder.external_show_hidden,
        file_count=file_count,
        latest_activity_at=latest_activity_at,
        **_folder_access_fields(index, folder, user),
        created_at=folder.created_at,
        updated_at=folder.updated_at,
    )


_README_BYTES_CAP = 512 * 1024  # 512 KiB — model descriptions don't need more
_README_PREFERRED_STEMS = ("readme", "description")


@router.get("/folders/{folder_id}/readme", response_model=FolderReadmeResponse)
async def get_folder_readme(
    folder_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Return the first markdown description file for a folder (#1268).

    Picks ``README.md`` / ``readme.md`` / ``description.md`` first (any case),
    otherwise the alphabetically-first ``*.md`` in the folder. 404 when no
    markdown file is present so the FE can hide the side panel.
    """
    user, can_read_all = auth_result

    await get_visible_folder(db, folder_id, user)

    query = LibraryFile.active().where(
        LibraryFile.folder_id == folder_id,
        func.lower(LibraryFile.filename).like("%.md"),
    )
    if user is not None and not can_read_all:
        query = query.where(LibraryFile.created_by_id == user.id)
    result = await db.execute(query)
    candidates = result.scalars().all()
    if not candidates:
        raise HTTPException(status_code=404, detail="No markdown description in folder")

    def sort_key(f: LibraryFile) -> tuple[int, str]:
        stem = os.path.splitext(f.filename.lower())[0]
        try:
            return (_README_PREFERRED_STEMS.index(stem), f.filename.lower())
        except ValueError:
            return (len(_README_PREFERRED_STEMS), f.filename.lower())

    pick = sorted(candidates, key=sort_key)[0]

    abs_path = to_absolute_path(pick.file_path)
    if not abs_path or not abs_path.exists():
        raise HTTPException(status_code=404, detail="Markdown file missing on disk")

    try:
        raw = abs_path.read_bytes()
    except OSError as e:
        logger.warning("Folder readme read failed for %s: %s", abs_path, e)
        raise HTTPException(status_code=500, detail="Could not read markdown file") from None

    truncated = len(raw) > _README_BYTES_CAP
    if truncated:
        raw = raw[:_README_BYTES_CAP]
    # `errors="replace"` so a single bad byte never blanks the panel.
    content = raw.decode("utf-8", errors="replace")

    return FolderReadmeResponse(filename=pick.filename, content=content, truncated=truncated)


@router.put("/folders/{folder_id}", response_model=FolderResponse)
async def update_folder(
    folder_id: int,
    data: FolderUpdate,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_UPDATE_ALL,
            Permission.LIBRARY_UPDATE_OWN,
        )
    ),
    printer_scope: PrinterScope = RequestPrinterScope,
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Update a folder.

    library:update_all changes any folder. With library:update_own a user may
    rename their own folders and move them into another folder they can
    write to (#3201); linking and sharing stay with library:update_all.
    """
    user, can_modify_all = auth_result
    # library:update_all reaches every folder, seen or not, as before #3201.
    folder = await get_visible_folder(db, folder_id, None if can_modify_all else user)

    if not can_modify_all:
        if not can_rename_folder(folder, user):
            raise HTTPException(status_code=403, detail="Only the folder's owner can change it")
        if data.project_id is not None or data.archive_id is not None:
            raise HTTPException(status_code=403, detail="Linking folders requires library:update_all")
        if data.shared is not None:
            raise HTTPException(status_code=403, detail="Sharing folders requires library:update_all")
        if data.parent_id:
            await get_writable_folder(db, data.parent_id, user)

    if data.shared is not None:
        folder.shared = data.shared

    if data.name is not None:
        folder.name = data.name

    if data.parent_id is not None:
        new_parent_id = data.parent_id or None  # 0 means move to root
        # An external folder mirrors a path on disk, so it stays where the scan
        # put it; sending its current parent back is not a move (#3305).
        if folder.is_external and new_parent_id != folder.parent_id:
            raise HTTPException(status_code=400, detail="External folders cannot be moved")

        # Prevent circular reference
        if data.parent_id == folder_id:
            raise HTTPException(status_code=400, detail="Folder cannot be its own parent")

        if new_parent_id is not None:
            parent = await db.get(LibraryFolder, new_parent_id)
            if parent is None:
                raise HTTPException(status_code=404, detail="Parent folder not found")
            if parent.is_external and new_parent_id != folder.parent_id:
                raise HTTPException(status_code=400, detail="Cannot move a folder into an external folder")

            # Check for circular reference in ancestors
            current_id = data.parent_id
            while current_id is not None:
                if current_id == folder_id:
                    raise HTTPException(status_code=400, detail="Cannot move folder into its own subtree")
                parent_result = await db.execute(select(LibraryFolder.parent_id).where(LibraryFolder.id == current_id))
                current_id = parent_result.scalar()

            folder.parent_id = data.parent_id
        else:
            folder.parent_id = None

    # Update project_id (0 to unlink)
    if data.project_id is not None:
        if data.project_id == 0:
            folder.project_id = None
        else:
            # Verify project exists
            project_result = await db.execute(select(Project).where(Project.id == data.project_id))
            if not project_result.scalar_one_or_none():
                raise HTTPException(status_code=404, detail="Project not found")
            folder.project_id = data.project_id

    # Update archive_id (0 to unlink)
    if data.archive_id is not None:
        if data.archive_id == 0:
            # A link the caller can't see is not theirs to remove.
            if folder.archive_id in await _linkable_archives(db, [folder.archive_id], printer_scope, actor):
                folder.archive_id = None
        else:
            # Verify archive exists
            if data.archive_id not in await _linkable_archives(db, [data.archive_id], printer_scope, actor):
                raise HTTPException(status_code=404, detail="Archive not found")
            folder.archive_id = data.archive_id

    await db.commit()
    await db.refresh(folder)

    # Get file count + latest file activity (#1770) and names
    agg_result = await db.execute(
        select(
            func.count(LibraryFile.id),
            func.max(LibraryFile.updated_at),
        ).where(
            LibraryFile.folder_id == folder_id,
            LibraryFile.deleted_at.is_(None),
            *_own_files_filter(user),
        )
    )
    file_count, latest_file = agg_result.one()
    file_count = file_count or 0
    latest_activity_at = max(folder.updated_at, latest_file) if latest_file is not None else folder.updated_at

    # Get project and archive names
    project_name = None
    if folder.project_id:
        project_result = await db.execute(select(Project.name).where(Project.id == folder.project_id))
        project_name = project_result.scalar()
    archives = await _linkable_archives(db, [folder.archive_id], printer_scope, actor)

    return FolderResponse(
        id=folder.id,
        name=folder.name,
        parent_id=folder.parent_id,
        project_id=folder.project_id,
        project_name=project_name,
        **_archive_link(folder.archive_id, archives),
        is_external=folder.is_external,
        external_path=folder.external_path,
        external_readonly=folder.external_readonly,
        external_show_hidden=folder.external_show_hidden,
        file_count=file_count,
        latest_activity_at=latest_activity_at,
        **_folder_access_fields(await _load_index(db, user), folder, user),
        created_at=folder.created_at,
        updated_at=folder.updated_at,
    )


@router.delete("/folders/{folder_id}")
async def delete_folder(
    folder_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_DELETE_ALL,
            Permission.LIBRARY_DELETE_OWN,
        )
    ),
):
    """Delete a folder and all its contents (cascade).

    library:delete_all deletes any folder. With library:delete_own a user
    deletes their own folder when everything in it is theirs too (#3201), or
    an ownerless folder that is truly empty (#1781). See
    ``library_folder_access.folder_delete_blocker``.
    """
    user, can_modify_all = auth_result
    # library:delete_all reaches every folder, seen or not, as before #3201.
    folder = await get_visible_folder(db, folder_id, None if can_modify_all else user)

    if not can_modify_all:
        blocker = folder_delete_blocker(await load_folder_index(db), folder, user)
        if blocker:
            raise HTTPException(status_code=403, detail=blocker)

    # External folders: only remove DB records, never delete files from external path
    is_ext = folder.is_external

    # Get all files in this folder and subfolders to delete from disk
    async def get_all_file_ids(fid: int) -> list[int]:
        """Recursively get all file IDs in a folder tree."""
        file_ids = []

        # Get files in this folder
        files_result = await db.execute(
            select(LibraryFile.id, LibraryFile.file_path, LibraryFile.thumbnail_path, LibraryFile.is_external).where(
                LibraryFile.folder_id == fid
            )
        )
        for fid_val, file_path, thumb_path, file_is_ext in files_result.all():
            file_ids.append(fid_val)
            # Only delete non-external files from disk
            if not is_ext and not file_is_ext:
                try:
                    if file_path and os.path.exists(file_path):
                        os.remove(file_path)
                    if thumb_path and os.path.exists(thumb_path):
                        os.remove(thumb_path)
                except OSError as e:
                    logger.warning("Failed to delete file: %s", e)

        # Get child folders and recurse
        children_result = await db.execute(select(LibraryFolder.id).where(LibraryFolder.parent_id == fid))
        for (child_id,) in children_result.all():
            file_ids.extend(await get_all_file_ids(child_id))

        return file_ids

    doomed_file_ids = await get_all_file_ids(folder_id)

    # The folder cascade hard-deletes every file row under it, so the queue has
    # to be taken off them first — same as the single-file delete below (#2819).
    # The return value used to be discarded here, which is why this never
    # happened for a folder delete.
    from backend.app.services.library_trash import delete_dependent_variants, release_queue_references

    await delete_dependent_variants(db, doomed_file_ids)
    await release_queue_references(db, doomed_file_ids)
    for doomed_id in doomed_file_ids:
        remove_library_photos_dir(doomed_id)

    # Delete folder (cascade will handle files and subfolders)
    await db.delete(folder)
    await db.commit()

    return {"status": "success", "message": "Folder deleted"}


# ============ External Folder Endpoints ============

# GHSA-r2qv follow-up (audit finding I1): external-folder mount path uses an
# allowlist of operator-opted-in roots rather than the original denylist of
# system directories. The denylist shape was fail-open-on-growth — anything
# not enumerated (``/data`` containing other users' archives, ``/root``,
# arbitrary NFS/SMB mounts, the Bambuddy ``LOG_DIR``) could be mounted by any
# user with ``LIBRARY_UPLOAD``. The allowlist defaults to empty and is
# extended via the ``BAMBUDDY_EXTERNAL_ROOTS`` env var (colon-separated
# absolute paths). The route is additionally gated on ``SETTINGS_UPDATE``
# (admin scope) rather than ``LIBRARY_UPLOAD`` because mounting host paths
# is an operator-level capability that crosses user boundaries.


# Bambuddy-owned data directories. Hardcode-rejected even if the operator
# tries to add them to ``BAMBUDDY_EXTERNAL_ROOTS`` — mounting these would
# allow reading other users' archives, log files, or the static assets path.
def _bambuddy_reserved_roots() -> tuple[Path, ...]:
    """Resolved Bambuddy-owned directories that may NEVER be mounted as an
    external folder regardless of the operator's allowlist.

    Resolved at call time because tests patch ``settings.base_dir`` /
    ``settings.log_dir`` to a temp dir; resolving lazily picks up the
    patched values rather than module-import-time values.
    """
    from backend.app.core.config import settings as app_settings

    reserved = [app_settings.base_dir, app_settings.log_dir, app_settings.static_dir, app_settings.archive_dir]
    return tuple(Path(p).resolve() for p in reserved if p is not None)


def _allowed_external_roots() -> tuple[Path, ...]:
    """Parse ``BAMBUDDY_EXTERNAL_ROOTS`` into resolved allowed roots.

    Empty env var (the default) means external folders are disabled.
    Operators opt in explicitly: ``BAMBUDDY_EXTERNAL_ROOTS=/mnt/library:/srv/3d``
    Returns a tuple of resolved ``Path`` objects; entries that don't
    resolve to absolute paths are silently dropped (operator error, not
    a security boundary). Resolved lazily so tests can monkeypatch.
    """
    raw = os.environ.get("BAMBUDDY_EXTERNAL_ROOTS", "")
    roots: list[Path] = []
    for entry in raw.split(":"):
        entry = entry.strip()
        if not entry:
            continue
        try:
            resolved = Path(entry).resolve()
        except (OSError, RuntimeError):  # noqa: BLE001 — operator config error, not a security boundary
            continue
        if resolved.is_absolute():
            roots.append(resolved)
    return tuple(roots)


def _path_within(child: Path, parent: Path) -> bool:
    """Return True if ``child`` is ``parent`` or any descendant.

    Uses ``Path.relative_to`` semantics (raises ``ValueError`` on miss)
    instead of string ``startswith``, which would falsely match
    ``/data-other`` against ``/data``. ``Path.is_relative_to`` is the
    sanctioned form on Python 3.9+; both are available here.
    """
    try:
        child.relative_to(parent)
    except ValueError:
        return False
    return True


# Supported file extensions for external folder scanning
_SCANNABLE_EXTENSIONS = {
    ".3mf",
    ".gcode",
    ".gcode.3mf",
    ".stl",
    ".obj",
    ".step",
    ".stp",
    ".png",
    ".jpg",
    ".jpeg",
    ".gif",
    ".webp",
    ".svg",
    ".md",
    # Documents that ship alongside a job folder and now have in-app
    # previews (#2976): drawings/datasheets and part lists.
    ".pdf",
    ".csv",
    ".xlsx",
    ".ods",
}


def _validate_external_path(path_str: str) -> Path:
    """Validate an external path is safe to mount.

    Allowlist semantics:
    1. Path must be absolute and resolve cleanly (symlink-escape rejected
       implicitly by the resolved-startswith check below).
    2. Path must fall under one of the roots enumerated in
       ``BAMBUDDY_EXTERNAL_ROOTS``; empty allowlist (the default)
       means external folders are not available on this deployment.
    3. Path must NOT fall under any Bambuddy-owned directory (``base_dir``,
       ``log_dir``, ``static_dir``, ``archive_dir``) — the reserved set
       takes precedence over the allowlist, so an operator who accidentally
       sets ``BAMBUDDY_EXTERNAL_ROOTS=/`` does not expose ``/data``.
    4. Existence + directory-type + readability gates remain.
    """
    path = Path(path_str).resolve()

    if not path.is_absolute():
        raise HTTPException(status_code=400, detail="Path must be absolute")

    allowed_roots = _allowed_external_roots()
    if not allowed_roots:
        raise HTTPException(
            status_code=400,
            detail=(
                "External folders are not enabled on this deployment. Ask the "
                "operator to set BAMBUDDY_EXTERNAL_ROOTS=<colon-separated paths>."
            ),
        )

    # Reserved (Bambuddy-owned) paths are rejected before the allowlist check
    # so an over-broad allowlist (e.g. operator set "/" for testing) cannot
    # expose Bambuddy's own data dir or log dir.
    for reserved in _bambuddy_reserved_roots():
        if _path_within(path, reserved):
            raise HTTPException(
                status_code=400,
                detail=f"Cannot mount Bambuddy-managed directory: {reserved}",
            )

    if not any(_path_within(path, root) for root in allowed_roots):
        raise HTTPException(
            status_code=400,
            detail=(
                f"Path '{path}' is not within an allowed external root. "
                f"Allowed roots: {', '.join(str(r) for r in allowed_roots)}"
            ),
        )

    if not path.exists():
        raise HTTPException(status_code=400, detail=f"Path does not exist: {path}")

    if not path.is_dir():
        raise HTTPException(status_code=400, detail=f"Path is not a directory: {path}")

    # Check readability
    if not os.access(path, os.R_OK):
        raise HTTPException(status_code=400, detail=f"Path is not readable: {path}")

    return path


@router.post("/folders/external", response_model=FolderResponse)
async def create_external_folder(
    data: ExternalFolderCreate,
    db: AsyncSession = Depends(get_db, scope="function"),
    # GHSA-r2qv follow-up (I1): elevated from LIBRARY_UPLOAD to SETTINGS_UPDATE.
    # Registering a host filesystem path as a Bambuddy library folder is an
    # operator-level capability that crosses user boundaries (one user's
    # registered external folder is visible to every other user via
    # /api/v1/library/folders). LIBRARY_UPLOAD was always the wrong scope —
    # SETTINGS_UPDATE is the admin-class gate that already protects every
    # other host-affecting setting (SMTP, LDAP, cloud, smart plugs).
    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.SETTINGS_UPDATE)),
):
    """Create an external folder that points to a host directory."""
    resolved = _validate_external_path(data.external_path)

    # Check no other external folder already points to this path
    existing = await db.execute(
        select(LibraryFolder).where(
            LibraryFolder.is_external.is_(True),
            LibraryFolder.external_path == str(resolved),
        )
    )
    if existing.scalar_one_or_none():
        raise HTTPException(status_code=409, detail="An external folder already exists for this path")

    # Verify parent exists if specified
    if data.parent_id is not None:
        parent_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == data.parent_id))
        if not parent_result.scalar_one_or_none():
            raise HTTPException(status_code=404, detail="Parent folder not found")

    folder = LibraryFolder(
        name=data.name,
        parent_id=data.parent_id,
        is_external=True,
        created_by_id=current_user.id if current_user else None,
        # A mount is everyone's, as before #3201; an admin can stop sharing it.
        shared=True,
        external_path=str(resolved),
        external_readonly=data.readonly,
        external_show_hidden=data.show_hidden,
    )
    db.add(folder)
    await db.commit()
    await db.refresh(folder)

    return FolderResponse(
        id=folder.id,
        name=folder.name,
        parent_id=folder.parent_id,
        project_id=None,
        archive_id=None,
        is_external=True,
        external_path=folder.external_path,
        external_readonly=folder.external_readonly,
        external_show_hidden=folder.external_show_hidden,
        file_count=0,
        # Newly-created external folder hasn't been scanned yet — fall back
        # to the folder's own updated_at (#1770).
        latest_activity_at=folder.updated_at,
        **_folder_access_fields(await _load_index(db, current_user), folder, current_user),
        created_at=folder.created_at,
        updated_at=folder.updated_at,
    )


def _mtime_to_datetime(mtime: float) -> datetime:
    """Convert an ``os.stat().st_mtime`` epoch value to a naive-UTC datetime (#2680).

    Naive UTC to match the other library timestamp columns (``created_at`` /
    ``updated_at`` are naive ``func.now()``), so activity comparisons never mix
    naive and aware values on either dialect.
    """
    return datetime.fromtimestamp(mtime, tz=timezone.utc).replace(tzinfo=None)


@router.post("/folders/{folder_id}/scan")
async def scan_external_folder(
    folder_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Scan an external folder and sync files to the database.

    Discovers new files, removes DB entries for deleted files.
    Does not copy files — stores the external path directly.
    """
    # A mount the user can't see is 404, like a missing one (#3201).
    folder = await get_visible_folder(db, folder_id, actor)
    if not folder.is_external or not folder.external_path:
        raise HTTPException(status_code=400, detail="Not an external folder")

    ext_path = Path(folder.external_path)
    if not ext_path.exists() or not ext_path.is_dir():
        raise HTTPException(status_code=400, detail=f"External path is not accessible: {folder.external_path}")

    # Collect all existing child external subfolder IDs (single query)
    all_folder_ids = [folder_id]
    child_result = await db.execute(
        select(LibraryFolder).where(
            LibraryFolder.is_external.is_(True),
            LibraryFolder.parent_id.isnot(None),
        )
    )
    all_child_folders = child_result.scalars().all()

    # Walk the parent chain to find all descendants of folder_id
    parent_to_children: dict[int, list] = {}
    for cf in all_child_folders:
        parent_to_children.setdefault(cf.parent_id, []).append(cf)

    queue = [folder_id]
    while queue:
        pid = queue.pop()
        for child in parent_to_children.get(pid, []):
            all_folder_ids.append(child.id)
            queue.append(child.id)

    # Get existing DB files across root and all subfolders
    existing_result = await db.execute(
        LibraryFile.active().where(
            LibraryFile.folder_id.in_(all_folder_ids),
            LibraryFile.is_external.is_(True),
        )
    )
    existing_files = {f.file_path: f for f in existing_result.scalars().all()}

    # Build folder cache: relative path -> folder_id (for resolving subfolders)
    # Pre-populate with existing child folders keyed by their external_path
    folder_cache: dict[str, int] = {"": folder_id}
    for fid in all_folder_ids:
        if fid == folder_id:
            continue
        # Find the child folder object
        for cf in all_child_folders:
            if cf.id == fid and cf.external_path:
                try:
                    rel = str(Path(cf.external_path).relative_to(ext_path))
                    if rel != ".":
                        folder_cache[rel] = cf.id
                except ValueError:
                    pass

    # Scan the directory
    added = 0
    removed = 0
    found_paths: set[str] = set()
    seen_rel_dirs: set[str] = set()
    # Real on-disk mtime per visited folder id (#2680), applied after the walk.
    folder_mtimes: dict[int, datetime] = {}

    for dirpath, dirnames, filenames in os.walk(ext_path):
        # Filter hidden directories unless configured
        if not folder.external_show_hidden:
            dirnames[:] = [d for d in dirnames if not d.startswith(".")]

        rel_dir = str(Path(dirpath).relative_to(ext_path))
        if rel_dir == ".":
            rel_dir = ""
        seen_rel_dirs.add(rel_dir)

        # Resolve or create subfolder chain for this directory
        if rel_dir and rel_dir not in folder_cache:
            parts = Path(rel_dir).parts
            current_path = ""
            current_parent = folder_id
            for part in parts:
                current_path = f"{current_path}/{part}".lstrip("/")
                if current_path in folder_cache:
                    current_parent = folder_cache[current_path]
                else:
                    existing_sub = await db.execute(
                        select(LibraryFolder).where(
                            LibraryFolder.name == part,
                            LibraryFolder.parent_id == current_parent,
                            LibraryFolder.is_external.is_(True),
                        )
                    )
                    existing_folder = existing_sub.scalar_one_or_none()
                    if existing_folder:
                        current_parent = existing_folder.id
                    else:
                        new_folder = LibraryFolder(
                            name=part,
                            parent_id=current_parent,
                            is_external=True,
                            external_path=str(
                                ext_path / current_path
                            ),  # SEC-PATH-OK: current_path built from Path(rel_dir).parts of an os.walk descent under ext_path
                            external_readonly=folder.external_readonly,
                            external_show_hidden=folder.external_show_hidden,
                            # A scanned subfolder belongs with the mount (#3201).
                            created_by_id=folder.created_by_id,
                            shared=folder.shared,
                        )
                        db.add(new_folder)
                        await db.flush()
                        current_parent = new_folder.id
                    folder_cache[current_path] = current_parent

        target_folder_id = folder_cache.get(rel_dir, folder_id)

        # Record this directory's own mtime (#2680). os.walk visits every
        # directory once, so this covers the root external folder and every
        # subfolder (existing or just created). Applied to the folder rows
        # after the walk completes.
        try:
            folder_mtimes[target_folder_id] = _mtime_to_datetime(os.stat(dirpath).st_mtime)
        except OSError:
            pass

        for filename in filenames:
            # Skip hidden files unless configured
            if not folder.external_show_hidden and filename.startswith("."):
                continue

            filepath = (
                Path(dirpath) / filename
            )  # SEC-PATH-OK: dirpath + filename from os.walk(ext_path); filesystem-discovered, not user input
            ext = filepath.suffix.lower()

            # Check for compound extensions like .gcode.3mf
            if ext not in _SCANNABLE_EXTENSIONS:
                # Check compound
                compound = "".join(filepath.suffixes[-2:]).lower() if len(filepath.suffixes) >= 2 else ""
                if compound not in _SCANNABLE_EXTENSIONS:
                    continue

            # Resolve symlinks and ensure still under external_path
            try:
                real_path = filepath.resolve()
                real_path.relative_to(ext_path.resolve())
            except (ValueError, OSError):
                continue  # Symlink escapes the external dir

            file_path_str = str(filepath)
            found_paths.add(file_path_str)

            if file_path_str in existing_files:
                # Already tracked — refresh its on-disk mtime (#2680) so a file
                # edited/replaced over the mount (samba, etc.) re-sorts correctly
                # and old rows scanned before this field existed get backfilled.
                tracked = existing_files[file_path_str]
                try:
                    fs_mtime = _mtime_to_datetime(filepath.stat().st_mtime)
                except OSError:
                    fs_mtime = None
                if fs_mtime is not None and tracked.fs_modified_at != fs_mtime:
                    tracked.fs_modified_at = fs_mtime
                continue

            # Get file info
            try:
                stat = filepath.stat()
            except OSError:
                continue

            # The zip is opened for the thumbnail immediately below either way,
            # so judging a `.3mf` on its contents rather than its name (#2993)
            # costs this scan nothing.
            file_type = classify_file_type(filename, filepath)

            # Extract thumbnail for 3mf files (including .gcode.3mf sliced
            # outputs — those are 3MF zips on disk and carry the same
            # thumbnail Metadata/plate_1.png the parser reads). Pre-#1600
            # the gate was `file_type == "3mf"` alone, so .gcode.3mf files
            # in external folders silently got no thumbnail.
            thumbnail_path = None
            file_metadata = None
            if file_type in ("3mf", "gcode.3mf"):
                try:
                    parser = ThreeMFParser(str(filepath))
                    raw_metadata = parser.parse()
                    if raw_metadata:
                        # Extract thumbnail before cleaning metadata
                        thumb_data = raw_metadata.get("_thumbnail_data")
                        thumbnail_ext = raw_metadata.get("_thumbnail_ext", ".png")
                        if thumb_data:
                            thumb_dir = get_library_thumbnails_dir()
                            thumb_filename = f"{uuid.uuid4().hex}{thumbnail_ext}"
                            thumb_full = (
                                thumb_dir / thumb_filename
                            )  # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + thumbnail_ext
                            thumb_full.write_bytes(thumb_data)
                            thumbnail_path = to_relative_path(thumb_full)

                        # Clean metadata - remove non-JSON-serializable data (bytes, etc.)
                        def clean_metadata(obj):
                            if isinstance(obj, dict):
                                return {
                                    k: clean_metadata(v)
                                    for k, v in obj.items()
                                    if not isinstance(v, bytes) and k not in ("_thumbnail_data", "_thumbnail_ext")
                                }
                            elif isinstance(obj, list):
                                return [clean_metadata(i) for i in obj if not isinstance(i, bytes)]
                            elif isinstance(obj, bytes):
                                return None
                            return obj

                        file_metadata = clean_metadata(raw_metadata)
                except Exception as e:
                    logger.debug("Failed to extract metadata from external 3mf %s: %s", filepath, e)

            # STL and PDF thumbnails are deferred to a background task spawned
            # after the scan's db.commit() — see _backfill_external_thumbnails.
            # Doing them inline would block the HTTP request for minutes on a
            # large NAS mount (#1299).

            # Extract gcode thumbnail
            if file_type == "gcode" and thumbnail_path is None:
                thumb_data = extract_gcode_thumbnail(filepath)
                if thumb_data:
                    thumb_dir = get_library_thumbnails_dir()
                    thumb_filename = f"{uuid.uuid4().hex}.png"
                    thumb_full = thumb_dir / thumb_filename  # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + ".png"
                    thumb_full.write_bytes(thumb_data)
                    thumbnail_path = to_relative_path(thumb_full)

            # Create thumbnail for image files
            if ext.lower() in IMAGE_EXTENSIONS and thumbnail_path is None:
                thumbnail_path_str = create_image_thumbnail(filepath, get_library_thumbnails_dir())
                if thumbnail_path_str:
                    thumbnail_path = to_relative_path(Path(thumbnail_path_str))

            db_file = LibraryFile(
                folder_id=target_folder_id,
                is_external=True,
                filename=filename,
                file_path=file_path_str,
                file_type=file_type,
                file_size=stat.st_size,
                file_hash=None,  # Skip hashing external files for performance
                thumbnail_path=thumbnail_path,
                file_metadata=_without_print_name(file_metadata),
                fs_modified_at=_mtime_to_datetime(stat.st_mtime),  # #2680: real on-disk mtime
            )
            db.add(db_file)
            added += 1

    # Remove DB entries for files that no longer exist on disk.
    #
    # Gate on actual disk presence, NOT merely absence from found_paths:
    # found_paths only collects extensions in _SCANNABLE_EXTENSIONS, so a
    # record for any other file the upload path admitted (e.g. a .md README,
    # #2520) would otherwise be treated as "deleted from disk" and purged on
    # every scan even though the file is still there. os.path.exists keeps
    # such records; genuinely-deleted files (absent from disk) are still
    # cleaned up. External file_path is the absolute on-disk path.
    vanished_ids: list[int] = []
    for path_str, db_file in existing_files.items():
        if path_str not in found_paths and not os.path.exists(path_str):
            # Clean up thumbnail if we generated one
            if db_file.thumbnail_path:
                try:
                    abs_thumb = to_absolute_path(db_file.thumbnail_path)
                    if abs_thumb and abs_thumb.exists():
                        abs_thumb.unlink()
                except OSError:
                    pass
            # The row is gone for good — external files skip the trash — so
            # its photos go with it rather than being orphaned under an id
            # nothing points at any more (#3077).
            remove_library_photos_dir(db_file.id)
            vanished_ids.append(db_file.id)
    if vanished_ids:
        # Same clean-up as every other hard delete of library rows, once for
        # all of them: a share that lost thousands of files must not run it
        # per file while holding the SQLite write lock.
        from backend.app.services.library_trash import delete_dependent_variants, release_queue_references

        await delete_dependent_variants(db, vanished_ids)
        await release_queue_references(db, vanished_ids)
        await db.execute(delete(LibraryFile).where(LibraryFile.id.in_(vanished_ids)))
        removed += len(vanished_ids)

    # Remove empty subfolders whose directories no longer exist on disk
    # Process deepest-first by sorting on path depth (descending)
    subfolder_entries = [(rel, fid) for rel, fid in folder_cache.items() if rel and fid != folder_id]
    subfolder_entries.sort(key=lambda x: x[0].count("/"), reverse=True)
    for rel_path, sub_fid in subfolder_entries:
        if rel_path in seen_rel_dirs:
            continue  # Directory still exists on disk
        # Check if subfolder has any remaining files
        file_count_result = await db.execute(
            select(func.count(LibraryFile.id)).where(
                LibraryFile.folder_id == sub_fid,
                LibraryFile.deleted_at.is_(None),
            )
        )
        if (file_count_result.scalar() or 0) == 0:
            # Check if it has any remaining child folders
            child_count_result = await db.execute(
                select(func.count(LibraryFolder.id)).where(LibraryFolder.parent_id == sub_fid)
            )
            if (child_count_result.scalar() or 0) == 0:
                sub_folder_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == sub_fid))
                sub_folder_obj = sub_folder_result.scalar_one_or_none()
                if sub_folder_obj:
                    await db.delete(sub_folder_obj)
                    folder_mtimes.pop(sub_fid, None)

    # Persist each visited folder's real directory mtime (#2680). Fetched in one
    # trip; folders deleted by the cleanup above were dropped from folder_mtimes.
    if folder_mtimes:
        folders_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id.in_(list(folder_mtimes.keys()))))
        for folder_obj in folders_result.scalars().all():
            new_mtime = folder_mtimes.get(folder_obj.id)
            if new_mtime is not None and folder_obj.fs_modified_at != new_mtime:
                folder_obj.fs_modified_at = new_mtime

    await db.commit()

    # Spawn STL/PDF thumbnail backfill in the background — the scan endpoint
    # returns immediately so the FE modal closes and subdirectories are
    # visible right away; thumbnails fill in over the following seconds /
    # minutes as the task processes each file. Survives FE refresh —
    # the task lives in the FastAPI event loop, not the request scope.
    # folder_cache.values() covers the root + every pre-existing subfolder
    # + every subfolder created during this scan. all_folder_ids on its own
    # would miss the newly-created ones (it's snapshotted before the walk).
    spawn_background_task(
        _backfill_external_thumbnails(list(set(folder_cache.values()))),
        name=f"thumbnail-backfill-folder-{folder_id}",
    )

    return {"status": "success", "added": added, "removed": removed}


# ============ File Endpoints ============


@router.get("/files", response_model=list[FileListResponse])
@router.get("/files/", response_model=list[FileListResponse])
async def list_files(
    response: Response,
    folder_id: int | None = None,
    project_id: int | None = None,
    include_root: bool = True,
    internal_only: bool = False,
    external_only: bool = False,
    recursive: bool = False,
    tag_ids: list[int] = Query(default_factory=list),
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """List files, optionally filtered by folder or project.

    Args:
        folder_id: Filter by folder ID. If None and include_root=True, returns root files.
        project_id: Return all files across folders linked to this project (bulk fetch, avoids N+1).
        include_root: If True and folder_id is None, returns files at root level.
                     If False and folder_id is None, returns all files.
        internal_only: Restrict the result to files in managed storage (`is_external=False`).
                       Used by the File Manager's "All Files" sidebar entry so a linked NAS
                       with hundreds of files doesn't drown the user's own uploads (#1621).
        external_only: Restrict the result to files under external folders
                       (`is_external=True`) — the symmetric combined view for users with
                       multiple linked external sources (#1621).
        recursive: When combined with ``folder_id``, also include files in every
                   descendant subfolder (#1268). Implemented via a recursive CTE
                   that walks ``library_folders.parent_id``. Default off so
                   existing callers (folder browsing, etc.) keep their narrow
                   single-folder semantics.
        tag_ids: Restrict the listing to files carrying ALL of these tags
                 (AND semantics, #1268). When non-empty the folder filter is
                 intentionally bypassed — tags are cross-cutting and the user
                 wants "every file with this tag" regardless of where it lives.
                 ``recursive`` becomes irrelevant in that case.
    """
    if internal_only and external_only:
        raise HTTPException(
            status_code=400,
            detail="internal_only and external_only are mutually exclusive",
        )

    user, can_read_all = auth_result
    query = LibraryFile.active().options(
        selectinload(LibraryFile.created_by),
        selectinload(LibraryFile.tags),
    )
    if user is not None and not can_read_all:
        query = query.where(LibraryFile.created_by_id == user.id)

    if tag_ids:
        # Cross-cutting filter — every requested tag must be present on the
        # file. JOIN + GROUP BY + HAVING COUNT(DISTINCT) is portable across
        # SQLite and Postgres without dialect tricks. We deliberately skip
        # the folder / project / include_root scoping below so the result
        # is the global "all files carrying these tags".
        unique_tag_ids = list(dict.fromkeys(tag_ids))
        query = (
            query.join(LibraryFileTag, LibraryFileTag.file_id == LibraryFile.id)
            .where(LibraryFileTag.tag_id.in_(unique_tag_ids))
            .group_by(LibraryFile.id)
            .having(func.count(distinct(LibraryFileTag.tag_id)) == len(unique_tag_ids))
        )
    elif folder_id is not None and recursive:
        # Walk the subtree starting at folder_id and collect every descendant
        # id. Recursive CTE works on both SQLite (>=3.8.3, shipped 2014) and
        # Postgres without dialect branching.
        roots = (
            select(LibraryFolder.id).where(LibraryFolder.id == folder_id).cte(name="folder_descendants", recursive=True)
        )
        descendants = roots.union_all(select(LibraryFolder.id).join(roots, LibraryFolder.parent_id == roots.c.id))
        query = query.where(LibraryFile.folder_id.in_(select(descendants.c.id)))
    elif folder_id is not None:
        query = query.where(LibraryFile.folder_id == folder_id)
    elif project_id is not None:
        # Single join instead of one query per folder (avoids N+1 pattern)
        query = query.join(LibraryFolder, LibraryFile.folder_id == LibraryFolder.id)
        query = query.where(LibraryFolder.project_id == project_id)
    elif include_root:
        query = query.where(LibraryFile.folder_id.is_(None))

    if internal_only:
        query = query.where(LibraryFile.is_external.is_(False))
    elif external_only:
        query = query.where(LibraryFile.is_external.is_(True))

    query = query.order_by(LibraryFile.filename)
    result = await db.execute(query)
    files = result.scalars().unique().all() if tag_ids else result.scalars().all()

    # Get duplicate counts
    hash_counts = {}
    if files:
        hashes = [f.file_hash for f in files if f.file_hash]
        if hashes:
            dup_result = await db.execute(
                select(LibraryFile.file_hash, func.count(LibraryFile.id))
                .where(LibraryFile.file_hash.in_(hashes), LibraryFile.deleted_at.is_(None))
                .group_by(LibraryFile.file_hash)
            )
            hash_counts = {h: c - 1 for h, c in dup_result.all()}  # -1 to exclude self

    # Variant group sizes (#671 / #2570). Counted across the whole group rather
    # than the rows on screen — members can sit in different folders, so counting
    # the listing would under-report and the "2 versions" badge would blink in
    # and out as the user navigated.
    variant_counts: dict[int, int] = {}
    group_ids = {f.variant_group_id for f in files if f.variant_group_id}
    if group_ids:
        count_result = await db.execute(
            select(LibraryFile.variant_group_id, func.count(LibraryFile.id))
            .where(LibraryFile.variant_group_id.in_(group_ids), LibraryFile.deleted_at.is_(None))
            .group_by(LibraryFile.variant_group_id)
        )
        variant_counts = dict(count_result.all())

    # Prevent browser caching of file list
    response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"

    file_list = []
    for f in files:
        # Extract key metadata for display
        print_name = None
        print_time = None
        filament_grams = None
        sliced_for_model = None
        if f.file_metadata:
            print_name = f.file_metadata.get("print_name")
            print_time = f.file_metadata.get("print_time_seconds")
            filament_grams = f.file_metadata.get("filament_used_grams")
            sliced_for_model = f.file_metadata.get("sliced_for_model")

        file_list.append(
            FileListResponse(
                id=f.id,
                folder_id=f.folder_id,
                is_external=f.is_external,
                filename=f.filename,
                file_type=f.file_type,
                file_size=f.file_size,
                thumbnail_path=f.thumbnail_path,
                print_count=f.print_count,
                duplicate_count=hash_counts.get(f.file_hash, 0) if f.file_hash else 0,
                created_by_id=f.created_by_id,
                created_by_username=f.created_by.username if f.created_by else None,
                created_at=f.created_at,
                fs_modified_at=f.fs_modified_at,
                print_name=print_name,
                print_time_seconds=print_time,
                filament_used_grams=filament_grams,
                sliced_for_model=sliced_for_model,
                tags=[TagSummary(id=t.id, name=t.name) for t in f.tags],
                variant_group_id=f.variant_group_id,
                variant_count=variant_counts.get(f.variant_group_id, 0) if f.variant_group_id else 0,
                external_url=f.external_url,
                has_notes=bool(f.notes),
                photo_count=len(f.photos or []),
            )
        )

    return file_list


@router.post("/files", response_model=FileUploadResponse)
@router.post("/files/", response_model=FileUploadResponse)
async def upload_file(
    file: UploadFile = File(...),
    folder_id: int | None = None,
    generate_stl_thumbnails: bool = Query(default=True),
    db: AsyncSession = Depends(get_db, scope="function"),
    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Upload a file to the library."""
    try:
        if not file.filename:
            raise HTTPException(status_code=400, detail="Filename is required")

        filename = file.filename
        # Reject FAT32/exFAT-incompatible filenames up front (#1540).
        try:
            validate_print_filename(filename)
        except InvalidFilenameError as e:
            raise HTTPException(status_code=400, detail=str(e)) from e
        # `ext` stays the trailing extension because the on-disk filename uses
        # it directly and the 3MF-parse branch below still gates on
        # `ext == ".3mf"`, which is correct for both `.3mf` and `.gcode.3mf`.
        # `file_type` is compound-aware and is decided further down, once the
        # bytes are on disk to be read.
        ext = os.path.splitext(filename)[1].lower()

        # Verify folder exists if specified, and that the user may add to it (#3201)
        target_folder = None
        if folder_id is not None:
            target_folder = await get_writable_folder(db, folder_id, actor)

        # Writable external folders write through to the mount so the file is
        # visible outside Bambuddy (#1112); everything else lands under the
        # internal library dir with a UUID-scoped filename. Resolved BEFORE
        # the content validation below so folder-permission rejections
        # (403 read-only, 400 missing path, 409 collision) still surface
        # before any "bad file format" 400 — preserves existing error
        # ordering / tests.
        file_path, is_external_upload = _resolve_upload_destination(target_folder, filename)

        # Read upload now so the validation can sniff magic bytes; the file
        # is written to disk only after the checks. #1401.
        content = await file.read()
        validate_print_file_upload(filename, content)

        # Save file
        with open(file_path, "wb") as f:
            f.write(content)

        # Now that the bytes are on disk the zip can settle what the name only
        # guessed at: a sliced 3MF uploaded as `Foo.3mf` is a sliced 3MF (#2993).
        file_type = classify_file_type(filename, file_path)

        # Calculate hash
        file_hash = calculate_file_hash(file_path)

        # Check for duplicates
        dup_result = await db.execute(
            select(LibraryFile.id).where(LibraryFile.file_hash == file_hash, LibraryFile.deleted_at.is_(None)).limit(1)
        )
        duplicate_of = dup_result.scalar()

        # Extract metadata and thumbnail
        metadata = {}
        thumbnail_path = None
        thumbnails_dir = get_library_thumbnails_dir()

        if ext == ".3mf":
            try:
                parser = ThreeMFParser(str(file_path))
                raw_metadata = parser.parse()

                # Extract thumbnail before cleaning metadata
                thumbnail_data = raw_metadata.get("_thumbnail_data")
                thumbnail_ext = raw_metadata.get("_thumbnail_ext", ".png")

                # Save thumbnail if extracted
                if thumbnail_data:
                    thumb_filename = f"{uuid.uuid4().hex}{thumbnail_ext}"
                    thumb_path = (
                        thumbnails_dir / thumb_filename
                    )  # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + thumbnail_ext
                    with open(thumb_path, "wb") as f:
                        f.write(thumbnail_data)
                    thumbnail_path = str(thumb_path)

                # Clean metadata - remove non-JSON-serializable data (bytes, etc.)
                def clean_metadata(obj):
                    if isinstance(obj, dict):
                        return {
                            k: clean_metadata(v)
                            for k, v in obj.items()
                            if not isinstance(v, bytes) and k not in ("_thumbnail_data", "_thumbnail_ext")
                        }
                    elif isinstance(obj, list):
                        return [clean_metadata(i) for i in obj if not isinstance(i, bytes)]
                    elif isinstance(obj, bytes):
                        return None
                    return obj

                metadata = clean_metadata(raw_metadata)
            except Exception as e:
                logger.warning("Failed to parse 3MF: %s", e)

        elif ext == ".gcode":
            # Extract embedded thumbnail from gcode
            try:
                thumbnail_data = extract_gcode_thumbnail(file_path)
                if thumbnail_data:
                    thumb_filename = f"{uuid.uuid4().hex}.png"
                    thumb_path = (
                        thumbnails_dir / thumb_filename
                    )  # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + ".png"
                    with open(thumb_path, "wb") as f:
                        f.write(thumbnail_data)
                    thumbnail_path = str(thumb_path)
            except Exception as e:
                logger.warning("Failed to extract gcode thumbnail: %s", e)

        elif ext.lower() in IMAGE_EXTENSIONS:
            # For image files, create a thumbnail from the image itself
            thumbnail_path = create_image_thumbnail(file_path, thumbnails_dir)

        elif ext.lower() == ".pdf":
            # Page one, rendered server-side; the browser preview's own
            # render (POST /preview-thumbnail) remains the fallback.
            thumbnail_path = await asyncio.to_thread(generate_pdf_thumbnail, file_path, thumbnails_dir)

        elif ext == ".stl":
            # Generate STL thumbnail if enabled. Same MIN_USABLE_STL_BYTES
            # pre-skip as extract_zip_file — stubs / placeholders below this
            # size can't contain a triangle so trimesh would return an empty
            # mesh anyway.
            if generate_stl_thumbnails:
                try:
                    if file_path.stat().st_size >= MIN_USABLE_STL_BYTES:
                        thumbnail_path = await _render_stl_thumbnail(file_path, thumbnails_dir)
                except OSError:
                    pass

        # Create database entry (managed files store relative paths for portability;
        # external files store the absolute mount path — same shape as scan produces)
        library_file = LibraryFile(
            folder_id=folder_id,
            is_external=is_external_upload,
            filename=filename,
            file_path=_stored_file_path(file_path, is_external_upload),
            file_type=file_type,
            file_size=len(content),
            file_hash=file_hash,
            thumbnail_path=to_relative_path(thumbnail_path) if thumbnail_path else None,
            file_metadata=_without_print_name(metadata) if metadata else None,
            created_by_id=actor.id if actor else None,
        )
        db.add(library_file)
        await db.commit()
        await db.refresh(library_file)

        return FileUploadResponse(
            id=library_file.id,
            filename=library_file.filename,
            file_type=library_file.file_type,
            file_size=library_file.file_size,
            thumbnail_path=library_file.thumbnail_path,
            duplicate_of=duplicate_of,
            metadata=library_file.file_metadata,
        )
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Upload failed for %s: %s", file.filename, e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}")


@router.post("/files/extract-zip", response_model=ZipExtractResponse)
async def extract_zip_file(
    file: UploadFile = File(...),
    folder_id: int | None = Query(default=None),
    preserve_structure: bool = Query(default=True),
    create_folder_from_zip: bool = Query(default=False),
    generate_stl_thumbnails: bool = Query(default=True),
    db: AsyncSession = Depends(get_db, scope="function"),
    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Upload and extract a ZIP file to the library.

    Args:
        file: The ZIP file to extract
        folder_id: Target folder ID (None = root)
        preserve_structure: If True, recreate folder structure from ZIP; if False, extract all files flat
        create_folder_from_zip: If True, create a folder named after the ZIP file and extract into it
        generate_stl_thumbnails: If True, generate thumbnails for STL files
    """
    import tempfile

    if not file.filename or not file.filename.lower().endswith(".zip"):
        raise HTTPException(status_code=400, detail="Only ZIP files are supported")

    # Verify target folder exists if specified, and that the user may add to it (#3201)
    if folder_id is not None:
        target_folder = await get_writable_folder(db, folder_id, actor)
        if target_folder.is_external and target_folder.external_readonly:
            raise HTTPException(status_code=403, detail="Cannot extract ZIP to a read-only external folder")
        if target_folder.is_external:
            # Writable external folders aren't supported by extract-zip because the
            # nested-subfolder creation path would need to mkdir on the mount and
            # create matching is_external=True LibraryFolder rows — a separate
            # design. Direct the user at Scan, which already handles that shape
            # (#1112).
            raise HTTPException(
                status_code=400,
                detail=(
                    "Cannot extract ZIP directly into an external folder. "
                    "Extract the ZIP on the external mount and run 'Scan External Folder' instead."
                ),
            )

    # Save ZIP to temp file
    try:
        with tempfile.NamedTemporaryFile(delete=False, suffix=".zip") as tmp:
            content = await file.read()
            tmp.write(content)
            tmp_path = tmp.name
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to save ZIP file: {str(e)}")

    extracted_files: list[ZipExtractResult] = []
    errors: list[ZipExtractError] = []
    folders_created = 0
    folder_cache: dict[str, int] = {}  # path -> folder_id

    # The upload may have taken a while; and again before each file below.
    await require_actor_still_exists(db, actor)

    # If create_folder_from_zip is True, create a folder named after the ZIP file
    zip_folder_id = folder_id
    logger.info(
        f"ZIP extraction: create_folder_from_zip={create_folder_from_zip}, folder_id={folder_id}, filename={file.filename}"
    )
    if create_folder_from_zip and file.filename:
        # Remove .zip extension to get folder name
        zip_folder_name = file.filename[:-4] if file.filename.lower().endswith(".zip") else file.filename
        # Check if folder already exists
        existing = await db.execute(
            select(LibraryFolder).where(
                LibraryFolder.name == zip_folder_name,
                LibraryFolder.parent_id == folder_id if folder_id else LibraryFolder.parent_id.is_(None),
            )
        )
        # Reuse a same-named folder only when the user may write to it; never
        # extract into someone else's folder that happens to share the name (#3201).
        existing_folder = next(
            (f for f in existing.scalars().all() if can_write_folder(f, actor)),
            None,
        )
        if existing_folder:
            zip_folder_id = existing_folder.id
            logger.info("Reusing existing folder '%s' with id=%s", zip_folder_name, zip_folder_id)
        else:
            # Create folder
            new_folder = LibraryFolder(
                name=zip_folder_name,
                parent_id=folder_id,
                created_by_id=actor.id if actor else None,
                # Made without a user (auth off, an API key): everyone's, as before #3201.
                shared=actor is None or actor.id is None,
            )
            db.add(new_folder)
            await db.flush()
            await db.commit()  # Commit folder creation immediately
            zip_folder_id = new_folder.id
            folders_created += 1
            logger.info("Created new folder '%s' with id=%s", zip_folder_name, zip_folder_id)

    try:
        with zipfile.ZipFile(tmp_path, "r") as zf:
            # Filter out directories and hidden/system files
            file_list = [
                name
                for name in zf.namelist()
                if not name.endswith("/")
                and not name.startswith("__MACOSX")
                and not os.path.basename(name).startswith(".")
            ]

            for zip_path in file_list:
                # An extract can run for minutes; a user deleted meanwhile must
                # not have the rest saved under their id (SQLite reuses it).
                await require_actor_still_exists(db, actor)
                try:
                    # Determine target folder (use zip_folder_id as base if create_folder_from_zip was used)
                    target_folder_id = zip_folder_id

                    if preserve_structure:
                        # Get directory path from ZIP
                        dir_path = os.path.dirname(zip_path)
                        if dir_path:
                            # Create folder structure
                            parts = dir_path.split("/")
                            current_parent = zip_folder_id
                            current_path = ""

                            for part in parts:
                                if not part:
                                    continue
                                current_path = f"{current_path}/{part}" if current_path else part

                                if current_path in folder_cache:
                                    current_parent = folder_cache[current_path]
                                else:
                                    # Check if folder exists
                                    existing = await db.execute(
                                        select(LibraryFolder).where(
                                            LibraryFolder.name == part,
                                            LibraryFolder.parent_id == current_parent
                                            if current_parent
                                            else LibraryFolder.parent_id.is_(None),
                                        )
                                    )
                                    existing_folder = next(
                                        (f for f in existing.scalars().all() if can_write_folder(f, actor)),
                                        None,
                                    )

                                    if existing_folder:
                                        current_parent = existing_folder.id
                                    else:
                                        # Create folder
                                        new_folder = LibraryFolder(
                                            name=part,
                                            parent_id=current_parent,
                                            created_by_id=actor.id if actor else None,
                                            # Made without a user (auth off, an API key): everyone's, as before #3201.
                                            shared=actor is None or actor.id is None,
                                        )
                                        db.add(new_folder)
                                        await db.flush()
                                        current_parent = new_folder.id
                                        folders_created += 1

                                    folder_cache[current_path] = current_parent

                            target_folder_id = current_parent

                    # Extract file
                    filename = os.path.basename(zip_path)
                    ext = os.path.splitext(filename)[1].lower()

                    # Generate unique filename for storage
                    unique_filename = f"{uuid.uuid4().hex}{ext}"
                    file_path = (
                        get_library_files_dir() / unique_filename
                    )  # SEC-PATH-OK: unique_filename = uuid.uuid4().hex + ext

                    # Extract and save file
                    file_content = zf.read(zip_path)
                    with open(file_path, "wb") as f:
                        f.write(file_content)

                    # Classified once the bytes are on disk so a sliced 3MF
                    # named `Foo.3mf` inside the zip is recognised as sliced
                    # (#2993) rather than trusted to say so in its name.
                    file_type = classify_file_type(filename, file_path)

                    # Calculate hash
                    file_hash = calculate_file_hash(file_path)

                    # Extract metadata and thumbnail for 3MF files
                    metadata = {}
                    thumbnail_path = None
                    thumbnails_dir = get_library_thumbnails_dir()

                    if ext == ".3mf":
                        try:
                            parser = ThreeMFParser(str(file_path))
                            raw_metadata = parser.parse()

                            thumbnail_data = raw_metadata.get("_thumbnail_data")
                            thumbnail_ext = raw_metadata.get("_thumbnail_ext", ".png")

                            if thumbnail_data:
                                thumb_filename = f"{uuid.uuid4().hex}{thumbnail_ext}"
                                thumb_path = (
                                    thumbnails_dir / thumb_filename
                                )  # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + thumbnail_ext
                                with open(thumb_path, "wb") as f:
                                    f.write(thumbnail_data)
                                thumbnail_path = str(thumb_path)

                            def clean_metadata(obj):
                                if isinstance(obj, dict):
                                    return {
                                        k: clean_metadata(v)
                                        for k, v in obj.items()
                                        if not isinstance(v, bytes) and k not in ("_thumbnail_data", "_thumbnail_ext")
                                    }
                                elif isinstance(obj, list):
                                    return [clean_metadata(i) for i in obj if not isinstance(i, bytes)]
                                elif isinstance(obj, bytes):
                                    return None
                                return obj

                            metadata = clean_metadata(raw_metadata)
                        except Exception as e:
                            logger.warning("Failed to parse 3MF from ZIP: %s", e)

                    elif ext == ".gcode":
                        try:
                            thumbnail_data = extract_gcode_thumbnail(file_path)
                            if thumbnail_data:
                                thumb_filename = f"{uuid.uuid4().hex}.png"
                                thumb_path = (
                                    thumbnails_dir / thumb_filename
                                )  # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + ".png"
                                with open(thumb_path, "wb") as f:
                                    f.write(thumbnail_data)
                                thumbnail_path = str(thumb_path)
                        except Exception as e:
                            logger.warning("Failed to extract gcode thumbnail from ZIP: %s", e)

                    elif ext.lower() in IMAGE_EXTENSIONS:
                        thumbnail_path = create_image_thumbnail(file_path, thumbnails_dir)

                    elif ext.lower() == ".pdf":
                        thumbnail_path = await asyncio.to_thread(generate_pdf_thumbnail, file_path, thumbnails_dir)

                    elif ext == ".stl":
                        # Generate STL thumbnail if enabled. Pre-skip files
                        # below MIN_USABLE_STL_BYTES — they can't contain
                        # even a single triangle, and bulk-uploaded ZIPs of
                        # stub STLs would otherwise log one debug line per
                        # file via the empty-mesh branch in trimesh.load.
                        if generate_stl_thumbnails and len(file_content) >= MIN_USABLE_STL_BYTES:
                            thumbnail_path = await _render_stl_thumbnail(file_path, thumbnails_dir)

                    # Create database entry (store relative paths for portability)
                    library_file = LibraryFile(
                        folder_id=target_folder_id,
                        filename=filename,
                        file_path=to_relative_path(file_path),
                        file_type=file_type,
                        file_size=len(file_content),
                        file_hash=file_hash,
                        thumbnail_path=to_relative_path(thumbnail_path) if thumbnail_path else None,
                        file_metadata=_without_print_name(metadata) if metadata else None,
                        created_by_id=actor.id if actor else None,
                    )
                    db.add(library_file)
                    await db.flush()
                    await db.refresh(library_file)

                    extracted_files.append(
                        ZipExtractResult(
                            filename=filename,
                            file_id=library_file.id,
                            folder_id=target_folder_id,
                        )
                    )

                    # Commit after each file to release database lock
                    # This prevents long-running transactions from blocking other requests
                    await db.commit()

                except Exception as e:
                    logger.error("Failed to extract %s: %s", zip_path, e)
                    errors.append(ZipExtractError(filename=os.path.basename(zip_path), error=str(e)))
                    # Rollback the failed file but continue with others
                    await db.rollback()

        return ZipExtractResponse(
            extracted=len(extracted_files),
            folders_created=folders_created,
            files=extracted_files,
            errors=errors,
        )

    except zipfile.BadZipFile:
        raise HTTPException(status_code=400, detail="Invalid or corrupted ZIP file")
    except HTTPException:
        raise
    except Exception as e:
        logger.error("ZIP extraction failed: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail=f"ZIP extraction failed: {str(e)}")
    finally:
        # Clean up temp file
        try:
            os.unlink(tmp_path)
        except OSError:
            pass  # Best-effort temp file cleanup; ignore if already removed


# ============ STL Thumbnail Batch Generation ============


@router.post("/generate-stl-thumbnails", response_model=BatchThumbnailResponse)
async def batch_generate_stl_thumbnails(
    request: BatchThumbnailRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_UPDATE_ALL,
            Permission.LIBRARY_UPDATE_OWN,
        )
    ),
):
    """Generate thumbnails for STL and PDF files in batch.

    With library:update_all this covers every matching file; with only
    library:update_own it is narrowed to the caller's own files, the same
    rule as update_file. The File Manager offers the toolbar button and the
    per-file "Generate Thumbnail" entry to update_own users, and both land
    here.

    PDFs are included so the ones added before server-side PDF thumbnails
    existed can be backfilled without opening each preview. The route keeps
    its name for API compatibility.

    Can generate thumbnails for:
    - Specific file IDs (file_ids)
    - All STL/PDF files in a folder (folder_id)
    - All STL/PDF files missing thumbnails (all_missing=True)
    """
    thumbnails_dir = get_library_thumbnails_dir()
    results: list[BatchThumbnailResult] = []

    # Build query based on request
    query = LibraryFile.active().where(LibraryFile.file_type.in_(("stl", "pdf")))

    user, can_modify_all = auth_result
    if not can_modify_all:
        query = query.where(LibraryFile.created_by_id == user.id)

    if request.file_ids:
        # Specific files
        query = query.where(LibraryFile.id.in_(request.file_ids))
    elif request.folder_id is not None:
        # All STL/PDF files in a specific folder
        query = query.where(LibraryFile.folder_id == request.folder_id)
        if not request.all_missing:
            # If not specifically asking for missing thumbnails, get all
            pass
        else:
            query = query.where(LibraryFile.thumbnail_path.is_(None))
    elif request.all_missing:
        # All STL/PDF files without thumbnails
        query = query.where(LibraryFile.thumbnail_path.is_(None))
    else:
        # No criteria specified - return empty
        return BatchThumbnailResponse(
            processed=0,
            succeeded=0,
            failed=0,
            results=[],
        )

    result = await db.execute(query)
    target_files = result.scalars().all()

    succeeded = 0
    failed = 0

    for target_file in target_files:
        file_path = to_absolute_path(target_file.file_path)

        if not file_path or not file_path.exists():
            results.append(
                BatchThumbnailResult(
                    file_id=target_file.id,
                    filename=target_file.filename,
                    success=False,
                    error="File not found on disk",
                )
            )
            failed += 1
            continue

        try:
            if target_file.file_type == "pdf":
                thumbnail_path = await asyncio.to_thread(generate_pdf_thumbnail, file_path, thumbnails_dir)
            else:
                thumbnail_path = await _render_stl_thumbnail(file_path, thumbnails_dir)

            if thumbnail_path:
                # Update database with relative path
                target_file.thumbnail_path = to_relative_path(thumbnail_path)
                await db.flush()
                results.append(
                    BatchThumbnailResult(
                        file_id=target_file.id,
                        filename=target_file.filename,
                        success=True,
                    )
                )
                succeeded += 1
            else:
                results.append(
                    BatchThumbnailResult(
                        file_id=target_file.id,
                        filename=target_file.filename,
                        success=False,
                        error="Thumbnail generation failed",
                    )
                )
                failed += 1
        except Exception as e:
            logger.error("Failed to generate thumbnail for %s: %s", target_file.filename, e)
            results.append(
                BatchThumbnailResult(
                    file_id=target_file.id,
                    filename=target_file.filename,
                    success=False,
                    error=str(e),
                )
            )
            failed += 1

    await db.commit()

    return BatchThumbnailResponse(
        processed=len(target_files),
        succeeded=succeeded,
        failed=failed,
        results=results,
    )


# ============ Queue Operations ============
# NOTE: These routes must be defined BEFORE /files/{file_id} to avoid path parameter conflicts


def is_sliced_file(filename: str) -> bool:
    """Check if a file is a sliced (printable) file.

    Sliced files are:
    - .gcode files
    - .3mf files that contain '.gcode.' in the name (e.g., filename.gcode.3mf)
    """
    lower = filename.lower()
    return lower.endswith(".gcode") or ".gcode." in lower


@router.post("/files/combine", response_model=FileUploadResponse)
async def combine_files(
    request: CombineFilesRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Combine STL library files into one multi-object 3MF.

    The slicer sidecar takes one model per slice, so putting several separate
    STLs (or several copies of one, #2999) on one plate means building that
    file first. The result is a new library file that slices like any other
    3MF; with auto-arrange on, the slicer lays the objects out on the bed.
    The sources are left untouched.
    """
    from backend.app.services.mesh_combine import CombinePart, MeshCombineError, combine_parts_to_3mf

    filename = request.filename.strip()
    if not filename.lower().endswith(".3mf"):
        filename = f"{filename}.3mf"
    try:
        validate_print_filename(filename)
    except InvalidFilenameError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e

    if request.folder_id is not None:
        await get_writable_folder(db, request.folder_id, actor)

    # Same per-row visibility the slice route applies: a READ_OWN caller must
    # not be able to pull another user's model into their own file by raw id.
    can_read_all = actor is None or actor.has_permission(Permission.LIBRARY_READ_ALL.value)

    # The same file listed twice is one object with the copies added up, so
    # its mesh is loaded and stored once. Order follows first appearance.
    copies_by_id: dict[int, int] = {}
    for item in request.items:
        copies_by_id[item.file_id] = copies_by_id.get(item.file_id, 0) + item.copies

    rows = (await db.execute(LibraryFile.active().where(LibraryFile.id.in_(copies_by_id)))).scalars().all()
    by_id = {row.id: row for row in rows}

    # Gate every source before touching any of them on disk, so the answer for
    # a file the caller can't see is the same 404 whatever else is in the list.
    sources = [_ensure_library_file_visible(by_id.get(file_id), actor, can_read_all) for file_id in copies_by_id]

    parts: list[CombinePart] = []
    for lib_file in sources:
        if not lib_file.filename.lower().endswith(".stl"):
            raise HTTPException(status_code=400, detail=f"Only STL files can be combined: {lib_file.filename}")
        src_path = _resolve_source_disk_path(lib_file)
        if src_path is None or not src_path.exists():
            raise HTTPException(status_code=404, detail=f"Source file missing on disk: {lib_file.filename}")
        parts.append(CombinePart(name=lib_file.filename, path=src_path, copies=copies_by_id[lib_file.id]))

    try:
        content = await asyncio.to_thread(combine_parts_to_3mf, parts)
    except MeshCombineError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e

    # The preview is embedded as Metadata/thumbnail.png by combine_parts_to_3mf,
    # so ThreeMFParser picks it up here like any other 3MF's thumbnail. Loading
    # the combined file back to render one would expand every copy.
    await require_actor_still_exists(db, actor)  # the combine may have taken a while
    library_file, _ = await save_3mf_bytes_to_library(
        db,
        file_bytes=content,
        filename=filename,
        folder_id=request.folder_id,
        source_type="combined",
        owner_id=actor.id if actor else None,
    )

    return FileUploadResponse(
        id=library_file.id,
        filename=library_file.filename,
        file_type=library_file.file_type,
        file_size=library_file.file_size,
        thumbnail_path=library_file.thumbnail_path,
        metadata=library_file.file_metadata,
    )


@router.post("/files/add-to-queue", response_model=AddToQueueResponse)
async def add_files_to_queue(
    request: AddToQueueRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.QUEUE_CREATE)),
    printer_scope: PrinterScope = RequestPrinterScope,
    review_required: bool = QueueReviewRequired,
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Add library files to the print queue.

    Only sliced files (.gcode or .gcode.3mf) can be added to the queue.
    The archive will be created automatically when the print starts.

    A caller may name a printer or a target model for the whole batch; with
    neither, each file is aimed at the model it says it was sliced for. The
    gates are the ones ``POST /queue/`` applies to a single item, because an
    item that reaches the scheduler through this route has to be as printable
    as one that reaches it through that one (#3112).
    """
    added: list[AddToQueueResult] = []
    errors: list[AddToQueueError] = []

    # Batch-level targeting. Rejected outright rather than per file: the whole
    # request names one destination, so a bad one is not a property of any
    # single file and reporting it fourteen times would say nothing extra.
    target_model_norm = normalize_model_name(request.target_model)
    if request.printer_id is not None and target_model_norm:
        raise HTTPException(400, "Cannot specify both printer_id and target_model")

    if request.printer_id is not None:
        printer_scope.ensure(request.printer_id)
        printer_row = (await db.execute(select(Printer).where(Printer.id == request.printer_id))).scalar_one_or_none()
        if not printer_row:
            raise HTTPException(400, "Printer not found")
    else:
        # Without a printer, every file goes to "any printer of a model"
        ensure_model_target_allowed(current_user, printer_scope)

    # Active printers of every model, read once, and only when the batch has no
    # printer of its own -- with one named, neither the check below nor the
    # inference in the loop consults it. The explicit target is validated for
    # the same reason POST /queue/ validates: a model nobody owns is a queue
    # item that waits forever. The inferred target reads the same set and
    # silently declines when it finds nothing, because there, owning no such
    # printer is the user's situation rather than their mistake -- the file
    # still queues, as the unassigned row it has always been.
    # Keyed without regard to case, as the scheduler matches a model, to the
    # printers' own spelling, which the items store.
    active_models: dict[str, str] = {}
    if request.printer_id is None:
        for (model,) in (
            await db.execute(select(Printer.model).where(Printer.is_active == True).order_by(Printer.id))  # noqa: E712
        ).all():
            if model:
                active_models.setdefault(model.lower(), model)
        if target_model_norm:
            if target_model_norm.lower() not in active_models:
                raise HTTPException(400, f"No active printers for model: {target_model_norm}")
            target_model_norm = active_models[target_model_norm.lower()]

    # Get all requested files
    result = await db.execute(LibraryFile.active().where(LibraryFile.id.in_(request.file_ids)))
    files = {f.id: f for f in result.scalars().all()}

    # Ownership-scoped reads apply here as everywhere else in this module: a
    # file the caller may not read is a file they may not print. Dropped from
    # the map rather than refused by name, so the per-file error below is the
    # same "File not found" an unknown id gets and the response says nothing
    # about which ids exist. Ownerless rows need LIBRARY_READ_ALL, matching
    # _ensure_library_file_visible.
    if actor is not None and not actor.has_permission(Permission.LIBRARY_READ_ALL.value):
        files = {fid: f for fid, f in files.items() if f.created_by_id is not None and f.created_by_id == actor.id}

    # Project attribution (#1897): a file queued from a project-linked folder
    # inherits that project, so the resulting archive counts toward the
    # project's progress. A file's own project link wins over its folder's.
    folder_ids = {f.folder_id for f in files.values() if f.folder_id is not None}
    folder_projects: dict[int, int | None] = {}
    if folder_ids:
        folder_result = await db.execute(
            select(LibraryFolder.id, LibraryFolder.project_id).where(LibraryFolder.id.in_(folder_ids))
        )
        folder_projects = dict(folder_result.all())

    # Get max position for queue ordering
    pos_result = await db.execute(select(func.coalesce(func.max(PrintQueueItem.position), 0)))
    max_position = pos_result.scalar() or 0

    # There is no per-job ask-for-outcome toggle on a bulk add, so the rows take
    # the same default the print dialog seeds itself from (#1898).
    confirm_outcome = await confirm_outcome_for_new_queue_item(db)
    # No print options on a bulk add either: the install's defaults, as the
    # print dialog would have offered them.
    print_options = await default_print_options(db)

    for file_id in request.file_ids:
        lib_file = files.get(file_id)

        if not lib_file:
            errors.append(AddToQueueError(file_id=file_id, filename="(not found)", error="File not found"))
            continue

        # A 3MF is judged by its content below, once it is known to be there;
        # anything else by its name, as before.
        lower_name = lib_file.filename.lower()
        if not lower_name.endswith(".3mf") and not is_sliced_file(lib_file.filename):
            errors.append(
                AddToQueueError(
                    file_id=file_id,
                    filename=lib_file.filename,
                    error="Not a sliced file. Only .gcode or .gcode.3mf files can be printed.",
                )
            )
            continue

        try:
            # Verify file exists on disk
            file_path = Path(app_settings.base_dir) / lib_file.file_path

            if not file_path.exists():
                errors.append(
                    AddToQueueError(file_id=file_id, filename=lib_file.filename, error="File not found on disk")
                )
                continue

            # Sliced is what the file holds, not what it is called (#3435): a
            # project saved as "Foo.gcode.3mf" holds no G-code, and a sliced
            # plate exported as "Foo.3mf" does (#2993).
            if lower_name.endswith(".3mf") and is_unsliced_3mf(file_path):
                errors.append(
                    AddToQueueError(
                        file_id=file_id, filename=lib_file.filename, error=unsliced_file_message(lib_file.filename)
                    )
                )
                continue
            # A zip that cannot be read has said nothing; its name decides, as before.
            if lower_name.endswith(".3mf") and not carries_gcode(file_path) and not is_sliced_file(lib_file.filename):
                errors.append(
                    AddToQueueError(
                        file_id=file_id,
                        filename=lib_file.filename,
                        error="Not a sliced file. Only .gcode or .gcode.3mf files can be printed.",
                    )
                )
                continue

            # The Bambu SD card is FAT32/exFAT, so an illegal character 553s at
            # upload time. POST /queue/ rejects those at queue time (#1540) and
            # this route did not, which turned a nameable mistake into a print
            # that failed hours later.
            try:
                validate_print_filename(lib_file.filename)
            except InvalidFilenameError as e:
                errors.append(AddToQueueError(file_id=file_id, filename=lib_file.filename, error=str(e)))
                continue

            # Where this file is aimed. An explicit printer wins; an explicit
            # model applies to every file and has to be one this file can
            # legally run on; with neither, the file's own declaration is used
            # when some active printer answers to it.
            item_printer_id = request.printer_id
            item_target_model: str | None = None
            if item_printer_id is None:
                if target_model_norm:
                    sliced_for = (lib_file.file_metadata or {}).get("sliced_for_model")
                    if not is_gcode_compatible(sliced_for, target_model_norm):
                        errors.append(
                            AddToQueueError(
                                file_id=file_id,
                                filename=lib_file.filename,
                                error=(
                                    f"File was sliced for {sliced_for} and cannot be dispatched to "
                                    f"{target_model_norm} printers"
                                ),
                            )
                        )
                        continue
                    item_target_model = target_model_norm
                else:
                    inferred = resolve_variant_model(lib_file)
                    item_target_model = active_models.get(inferred.lower()) if inferred else None

            # Filament the scheduler must match before handing a model-based
            # item to hardware. Without it the item goes to whichever printer
            # of that model is idle, whatever is loaded in it.
            # The plate that prints: added without one, a multi-plate file
            # prints its default plate. Stored on the item, so the filament a
            # printer must have, the AMS mapping and the unmatched-filament
            # check all read that plate rather than every plate.
            printed_plate = await asyncio.to_thread(printed_plate_of_multi_plate_file, file_path)
            required_filament_types = None
            if item_target_model:
                # POST /queue/'s own extractor, borrowed rather than
                # reimplemented: a second copy of this rule is a second thing
                # to keep in step.
                #
                # Off the loop, unlike there: that route parses one 3MF per
                # request and this one parses every file in the batch, so on a
                # bulk add of a few hundred -- especially from an external
                # folder on a NAS -- the zip reads add up to a stall the whole
                # event loop takes, status ingest included.
                filament_types = await asyncio.to_thread(_extract_filament_types_from_3mf, file_path, printed_plate)
                if filament_types:
                    required_filament_types = json.dumps(filament_types)

            # Create queue item referencing library file (archive created at print start)
            max_position += 1
            queue_item = PrintQueueItem(
                printer_id=item_printer_id,
                target_model=item_target_model,
                required_filament_types=required_filament_types,
                plate_id=printed_plate,
                library_file_id=file_id,
                project_id=lib_file.project_id
                or (folder_projects.get(lib_file.folder_id) if lib_file.folder_id is not None else None),
                position=max_position,
                status="pending",
                confirm_outcome=confirm_outcome,
                # Without this the row is ownerless, and `queue:read_own` filters
                # on `created_by_id` — so the user who queued the file could not
                # see it in their own queue.
                created_by_id=actor.id if actor else None,
                # Waits for someone to start it unless they may print without review (#1620)
                manual_start=review_required,
                **print_options,
            )
            db.add(queue_item)

            await db.flush()  # Get queue_item.id

            added.append(
                AddToQueueResult(
                    file_id=file_id,
                    filename=lib_file.filename,
                    queue_item_id=queue_item.id,
                )
            )

        except Exception as e:
            logger.exception("Error adding file %s to queue", file_id)
            errors.append(AddToQueueError(file_id=file_id, filename=lib_file.filename, error=str(e)))

    # Nothing queued and something to say about why. Returning 200 here is what
    # made this look like a working call that quietly did nothing: a client that
    # checks the status code sees success, and the reasons sit in a body it had
    # no cause to read (#3112). Partial success stays 200 -- items really were
    # created, and the per-file errors belong with them.
    if not added and errors:
        raise HTTPException(
            400,
            detail={
                "message": "No files could be added to the queue.",
                "errors": [e.model_dump() for e in errors],
            },
        )

    await db.commit()
    await announce_queue_jobs_added(db, [a.queue_item_id for a in added])

    return AddToQueueResponse(added=added, errors=errors)


@router.get("/files/{file_id}/plates")
async def get_library_file_plates(
    file_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Get available plates from a multi-plate 3MF library file.

    Returns a list of plates with their index, name, thumbnail availability,
    and filament requirements. For single-plate exports, returns a single plate.
    """
    import json

    import defusedxml.ElementTree as ET

    user, can_read_all = auth_result
    # Get the library file
    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    lib_file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)

    if not lib_file:
        raise HTTPException(status_code=404, detail="File not found")

    file_path = Path(app_settings.base_dir) / lib_file.file_path
    if not file_path.exists():
        raise HTTPException(status_code=404, detail="File not found on disk")

    # Only 3MF files have plates
    if not lib_file.filename.lower().endswith(".3mf"):
        return {"file_id": file_id, "filename": lib_file.filename, "plates": [], "is_multi_plate": False}

    plates = []
    # Printer / process preset names the 3MF was prepared with — used by the
    # SliceModal to default its dropdowns (#1325). Initialised here so the
    # final return never raises NameError when the file isn't a valid zip.
    embedded_presets: dict[str, str | None] = {"printer": None, "process": None}
    # Process settings the designer changed away from the stock preset (#2622).
    # Offered in the SliceModal so a cross-printer re-slice can carry them
    # instead of silently losing them to the picked process profile.
    design_overrides: list[dict] = []

    try:
        with zipfile.ZipFile(file_path, "r") as zf:
            namelist = zf.namelist()
            embedded_presets = extract_embedded_presets_from_3mf(zf)
            if _PROJECT_SETTINGS_PATH in namelist:
                try:
                    design_overrides = [
                        o._asdict()
                        for o in overrides_from_config(json.loads(zf.read(_PROJECT_SETTINGS_PATH).decode("utf-8")))
                    ]
                except (ValueError, OSError, KeyError):
                    design_overrides = []

            # Find all plate gcode files to determine available plates
            gcode_files = [n for n in namelist if n.startswith("Metadata/plate_") and n.endswith(".gcode")]

            # If no gcode is present (source-only or unsliced), fall back to plate JSON/PNG
            plate_indices: list[int] = []
            if gcode_files:
                # Extract plate indices from gcode filenames
                for gf in gcode_files:
                    try:
                        plate_str = gf[15:-6]  # Remove "Metadata/plate_" and ".gcode"
                        plate_indices.append(int(plate_str))
                    except ValueError:
                        pass  # Skip gcode file with non-numeric plate index
            else:
                plate_json_files = [n for n in namelist if n.startswith("Metadata/plate_") and n.endswith(".json")]
                plate_png_files = [
                    n
                    for n in namelist
                    if n.startswith("Metadata/plate_")
                    and n.endswith(".png")
                    and "_small" not in n
                    and "no_light" not in n
                ]
                plate_name_candidates = plate_json_files + plate_png_files
                plate_re = re.compile(r"^Metadata/plate_(\d+)\.(json|png)$")
                seen_indices: set[int] = set()
                for name in plate_name_candidates:
                    match = plate_re.match(name)
                    if match:
                        try:
                            index = int(match.group(1))
                        except ValueError:
                            continue
                        if index in seen_indices:
                            continue
                        seen_indices.add(index)
                        plate_indices.append(index)

            if not plate_indices:
                # No plate metadata found
                return {"file_id": file_id, "filename": lib_file.filename, "plates": [], "is_multi_plate": False}

            plate_indices.sort()

            # Parse model_settings.config for plate names + object assignments
            plate_names = {}
            plate_object_ids: dict[int, list[str]] = {}
            object_names_by_id: dict[str, str] = {}
            if "Metadata/model_settings.config" in namelist:
                try:
                    model_content = zf.read("Metadata/model_settings.config").decode()
                    model_root = ET.fromstring(model_content)
                    for obj_elem in model_root.findall(".//object"):
                        obj_id = obj_elem.get("id")
                        if not obj_id:
                            continue
                        name_meta = obj_elem.find("metadata[@key='name']")
                        obj_name = name_meta.get("value") if name_meta is not None else None
                        if obj_name:
                            object_names_by_id[obj_id] = obj_name

                    for plate_elem in model_root.findall(".//plate"):
                        plater_id = None
                        plater_name = None
                        for meta in plate_elem.findall("metadata"):
                            key = meta.get("key")
                            value = meta.get("value")
                            if key == "plater_id" and value:
                                try:
                                    plater_id = int(value)
                                except ValueError:
                                    pass  # Ignore plate with non-numeric plater_id
                            elif key == "plater_name" and value:
                                plater_name = value.strip()
                        if plater_id is not None and plater_name:
                            plate_names[plater_id] = plater_name

                        if plater_id is not None:
                            for instance_elem in plate_elem.findall("model_instance"):
                                for inst_meta in instance_elem.findall("metadata"):
                                    if inst_meta.get("key") == "object_id":
                                        obj_id = inst_meta.get("value")
                                        if not obj_id:
                                            continue
                                        plate_object_ids.setdefault(plater_id, [])
                                        if obj_id not in plate_object_ids[plater_id]:
                                            plate_object_ids[plater_id].append(obj_id)
                except Exception:
                    pass  # model_settings.config is optional; skip if missing or malformed

            # Parse slice_info.config for plate metadata
            plate_metadata = {}
            if "Metadata/slice_info.config" in namelist:
                content = zf.read("Metadata/slice_info.config").decode()
                root = ET.fromstring(content)

                for plate_elem in root.findall(".//plate"):
                    plate_info = {"filaments": [], "prediction": None, "weight": None, "name": None, "objects": []}

                    plate_index = None
                    for meta in plate_elem.findall("metadata"):
                        key = meta.get("key")
                        value = meta.get("value")
                        if key == "index" and value:
                            try:
                                plate_index = int(value)
                            except ValueError:
                                pass  # Ignore plate with non-numeric index
                        elif key == "prediction" and value:
                            try:
                                plate_info["prediction"] = int(value)
                            except ValueError:
                                pass  # Leave prediction as None if not a valid integer
                        elif key == "weight" and value:
                            try:
                                plate_info["weight"] = float(value)
                            except ValueError:
                                pass  # Leave weight as None if not a valid number

                    # Get filaments used in this plate
                    for filament_elem in plate_elem.findall("filament"):
                        filament_id = filament_elem.get("id")
                        filament_type = filament_elem.get("type", "")
                        filament_color = filament_elem.get("color", "")
                        used_g = filament_elem.get("used_g", "0")
                        used_m = filament_elem.get("used_m", "0")

                        try:
                            used_grams = float(used_g)
                        except (ValueError, TypeError):
                            used_grams = 0

                        if used_grams > 0 and filament_id:
                            plate_info["filaments"].append(
                                {
                                    "slot_id": int(filament_id),
                                    "type": filament_type,
                                    "color": filament_color,
                                    # The preset the slicer assigned (GFA01, ...): names the product line
                                    "tray_info_idx": filament_elem.get("tray_info_idx", "") or "",
                                    "used_grams": round(used_grams, 1),
                                    "used_meters": float(used_m) if used_m else 0,
                                }
                            )

                    plate_info["filaments"].sort(key=lambda x: x["slot_id"])

                    # Collect object names
                    for obj_elem in plate_elem.findall("object"):
                        obj_name = obj_elem.get("name")
                        if obj_name and obj_name not in plate_info["objects"]:
                            plate_info["objects"].append(obj_name)

                    # Set plate name
                    if plate_index is not None:
                        custom_name = plate_names.get(plate_index)
                        if custom_name:
                            plate_info["name"] = custom_name
                        elif plate_info["objects"]:
                            plate_info["name"] = plate_info["objects"][0]
                        plate_metadata[plate_index] = plate_info

            # Parse plate_*.json for object lists when slice_info is missing
            plate_json_objects: dict[int, list[str]] = {}
            for name in namelist:
                match = re.match(r"^Metadata/plate_(\d+)\.json$", name)
                if not match:
                    continue
                try:
                    plate_index = int(match.group(1))
                except ValueError:
                    continue
                try:
                    payload = json.loads(zf.read(name).decode())
                    bbox_objects = payload.get("bbox_objects", [])
                    names: list[str] = []
                    for obj in bbox_objects:
                        obj_name = obj.get("name") if isinstance(obj, dict) else None
                        if obj_name and obj_name not in names:
                            names.append(obj_name)
                    if names:
                        plate_json_objects[plate_index] = names
                except Exception:
                    continue

            # Build plate list
            for idx in plate_indices:
                meta = plate_metadata.get(idx, {})
                has_thumbnail = f"Metadata/plate_{idx}.png" in namelist
                objects = meta.get("objects", [])
                if not objects:
                    objects = plate_json_objects.get(idx, [])
                if not objects and plate_object_ids.get(idx):
                    objects = [
                        object_names_by_id.get(obj_id, f"Object {obj_id}") for obj_id in plate_object_ids.get(idx, [])
                    ]

                plate_name = meta.get("name")
                if not plate_name:
                    plate_name = plate_names.get(idx)
                if not plate_name and objects:
                    plate_name = objects[0]

                plates.append(
                    {
                        "index": idx,
                        "name": plate_name,
                        "objects": objects,
                        "object_count": len(objects),
                        "has_thumbnail": has_thumbnail,
                        "thumbnail_url": f"/api/v1/library/files/{file_id}/plate-thumbnail/{idx}"
                        if has_thumbnail
                        else None,
                        "print_time_seconds": meta.get("prediction"),
                        "filament_used_grams": meta.get("weight"),
                        "filaments": meta.get("filaments", []),
                    }
                )

    except Exception as e:
        logger.warning("Failed to parse plates from library file %s: %s", file_id, e)

    return {
        "file_id": file_id,
        "filename": lib_file.filename,
        "plates": plates,
        "is_multi_plate": len(plates) > 1,
        "embedded_printer": embedded_presets["printer"],
        "embedded_process": embedded_presets["process"],
        "design_overrides": design_overrides,
    }


@router.get("/files/{file_id}/plate-thumbnail/{plate_index}")
async def get_library_file_plate_thumbnail(
    file_id: int,
    plate_index: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_media_token_ownership(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Get the thumbnail image for a specific plate from a library file.

    Ownership-gated on the same terms as the file itself (#3025).
    """
    from starlette.responses import Response

    user, can_read_all = auth_result
    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    lib_file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)

    file_path = Path(app_settings.base_dir) / lib_file.file_path
    if not file_path.exists():
        raise HTTPException(status_code=404, detail="File not found on disk")

    try:
        with zipfile.ZipFile(file_path, "r") as zf:
            thumb_path = f"Metadata/plate_{plate_index}.png"
            if thumb_path in zf.namelist():
                data = zf.read(thumb_path)
                return Response(content=data, media_type="image/png")
    except Exception:
        pass  # Archive unreadable or thumbnail missing; fall through to 404

    raise HTTPException(status_code=404, detail=f"Thumbnail for plate {plate_index} not found")


async def _try_preview_slice_filaments(
    db: AsyncSession,
    *,
    kind: str,
    source_id: int,
    plate_id: int,
    file_path: Path,
    request_id: str | None = None,
) -> list[dict] | None:
    """Run a preview slice via the user's configured sidecar. Same shape as
    the matching helper in archives.py — see that module for rationale.

    ``request_id``: when supplied, forwarded to the sidecar so the
    SliceModal's inline spinner + toast can poll the matching progress
    endpoint and show "Generating G-code (45%)" for the preview as well.
    """
    from backend.app.api.routes.settings import get_setting
    from backend.app.services.slice_preview import get_preview_filaments
    from backend.app.services.slicer_api import get_stall_timeout_seconds

    preferred = (await get_setting(db, "preferred_slicer")) or "bambu_studio"
    if preferred == "orcaslicer":
        configured = await get_setting(db, "orcaslicer_api_url")
        api_url = (configured or app_settings.slicer_api_url).strip()
    elif preferred == "bambu_studio":
        configured = await get_setting(db, "bambu_studio_api_url")
        api_url = (configured or app_settings.bambu_studio_api_url).strip()
    else:
        return None
    if not api_url:
        return None

    try:
        file_bytes = file_path.read_bytes()
    except OSError:
        return None
    return await get_preview_filaments(
        kind=kind,
        source_id=source_id,
        plate_id=plate_id,
        file_bytes=file_bytes,
        file_name=file_path.name,
        api_url=api_url,
        request_id=request_id,
        timeout_seconds=await get_stall_timeout_seconds(db),
    )


@router.get("/files/{file_id}/filament-requirements")
async def get_library_file_filament_requirements(
    file_id: int,
    plate_id: int | None = None,
    request_id: str | None = None,
    full_slots: bool = False,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Get filament requirements from a library file.

    Parses the 3MF file to extract filament slot IDs, types, colors, and usage.
    This enables AMS slot assignment when printing from the file manager.

    Args:
        file_id: The library file ID
        plate_id: Optional plate index to get filaments for a specific plate
        full_slots: Return one entry per *project* slot rather than only the
            slots the plate consumes. See :func:`_expand_to_project_slots`.
            Only the slice modal wants this; print-time AMS matching must keep
            the used-only list.
    """
    import defusedxml.ElementTree as ET

    user, can_read_all = auth_result
    # Get the library file
    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    lib_file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)

    # Get the full file path
    file_path = Path(app_settings.base_dir) / lib_file.file_path

    if not file_path.exists():
        raise HTTPException(status_code=404, detail="File not found on disk")

    # Only 3MF files have parseable filament info
    if not lib_file.filename.lower().endswith(".3mf"):
        return {"file_id": file_id, "filename": lib_file.filename, "plate_id": plate_id, "filaments": []}

    filaments = []

    try:
        with zipfile.ZipFile(file_path, "r") as zf:
            # Parse slice_info.config for filament requirements
            if "Metadata/slice_info.config" in zf.namelist():
                content = zf.read("Metadata/slice_info.config").decode()
                root = ET.fromstring(content)

                if plate_id is not None:
                    # Find filaments for specific plate
                    for plate_elem in root.findall(".//plate"):
                        # Check if this is the requested plate
                        plate_index = None
                        for meta in plate_elem.findall("metadata"):
                            if meta.get("key") == "index":
                                try:
                                    plate_index = int(meta.get("value", ""))
                                except ValueError:
                                    pass  # Skip plate with non-numeric index value
                                break

                        if plate_index == plate_id:
                            # Extract filaments from this plate
                            for filament_elem in plate_elem.findall("filament"):
                                filament_id = filament_elem.get("id")
                                filament_type = filament_elem.get("type", "")
                                filament_color = filament_elem.get("color", "")
                                used_g = filament_elem.get("used_g", "0")
                                used_m = filament_elem.get("used_m", "0")

                                tray_info_idx = filament_elem.get("tray_info_idx", "")

                                try:
                                    used_grams = float(used_g)
                                except (ValueError, TypeError):
                                    used_grams = 0

                                if used_grams > 0 and filament_id:
                                    filaments.append(
                                        {
                                            "slot_id": int(filament_id),
                                            "type": filament_type,
                                            "color": filament_color,
                                            "used_grams": round(used_grams, 1),
                                            "used_meters": float(used_m) if used_m else 0,
                                            "tray_info_idx": tray_info_idx,
                                            # Sliced output already pre-filtered by used_g>0,
                                            # so every entry that survives is in fact used by
                                            # this plate. Print-dispatch consumers ignore the
                                            # flag; SliceModal uses it to enable/disable rows.
                                            "used_in_plate": True,
                                        }
                                    )
                            break
                else:
                    # Extract all filaments with used_g > 0 (for single-plate or overview)
                    for filament_elem in root.findall(".//filament"):
                        filament_id = filament_elem.get("id")
                        filament_type = filament_elem.get("type", "")
                        filament_color = filament_elem.get("color", "")
                        used_g = filament_elem.get("used_g", "0")
                        used_m = filament_elem.get("used_m", "0")

                        tray_info_idx = filament_elem.get("tray_info_idx", "")

                        try:
                            used_grams = float(used_g)
                        except (ValueError, TypeError):
                            used_grams = 0

                        if used_grams > 0 and filament_id:
                            filaments.append(
                                {
                                    "slot_id": int(filament_id),
                                    "type": filament_type,
                                    "color": filament_color,
                                    "used_grams": round(used_grams, 1),
                                    "used_meters": float(used_m) if used_m else 0,
                                    "tray_info_idx": tray_info_idx,
                                    "used_in_plate": True,
                                }
                            )

            # Re-slicing a source that already carries slice_info (#2712).
            # The block above answers "what does this plate consume", which is
            # what print-time AMS matching needs. The slice modal needs "what
            # slots exist", because its list is positional and the CLI binds
            # entry N to slot N — so a source using only slot 4 handed the
            # user's single pick to slot 1 and sliced slot 4 with the source's
            # embedded default. Widen here rather than in the modal so the
            # print path keeps the narrow list it depends on.
            if full_slots and filaments:
                filaments = expand_to_project_slots(zf, filaments)

            # Unsliced project files: slice_info had no per-plate data.
            # Return the FULL project_settings.config AMS slot list so
            # the slicer CLI receives a profile for every project slot
            # (otherwise it silently fills the gap from embedded
            # defaults — surfaces as "I picked white but the print has
            # grey" because the source's grey support filament leaks
            # into the output). Use the preview slice to mark which
            # slots the picked plate actually consumes; the SliceModal
            # disables the unused rows so the user only interacts with
            # the dropdowns that matter, while the backend still has
            # the complete list to pass to the CLI.
            if not filaments:
                project_filaments = extract_project_filaments_from_3mf(zf)
                used_slot_ids: set[int] = set()
                if project_filaments and plate_id is not None:
                    preview = await _try_preview_slice_filaments(
                        db,
                        kind="library_file",
                        source_id=file_id,
                        plate_id=plate_id,
                        file_path=file_path,
                        request_id=request_id,
                    )
                    if preview is not None:
                        used_slot_ids = {f["slot_id"] for f in preview}
                # Default to "every slot is used" when preview-slice
                # didn't produce data: better to over-enable dropdowns
                # than under-enable and have the user unable to pick a
                # filament the plate actually uses.
                fallback_all_used = not used_slot_ids
                for f in project_filaments:
                    f["used_in_plate"] = fallback_all_used or f["slot_id"] in used_slot_ids
                filaments = project_filaments

            # Sort by slot ID
            filaments.sort(key=lambda x: x["slot_id"])

            # Enrich with nozzle mapping for dual-nozzle printers
            nozzle_mapping = extract_nozzle_mapping_from_3mf(zf, plate_id=plate_id)
            if nozzle_mapping:
                for filament in filaments:
                    filament["nozzle_id"] = nozzle_mapping.get(filament["slot_id"])

            # Nozzle-rack machines (#1784): the print dialog offers a rack
            # position per filament group, which needs the group table as well
            # as the carriage above.
            annotate_rack_groups(filaments, file_path, plate_id)

    except Exception as e:
        logger.warning("Failed to parse filament requirements from library file %s: %s", file_id, e)

    return {
        "file_id": file_id,
        "filename": lib_file.filename,
        "plate_id": plate_id,
        "filaments": filaments,
    }


def _patch_process_bed_type(process_json: str, bed_type: str) -> str:
    """Overwrite ``curr_bed_type`` in a process-profile JSON before forwarding
    to the slicer sidecar.

    The slicer CLI reads the build-plate type from the process profile's
    ``curr_bed_type`` field. When the user picks a non-default plate in the
    SliceModal (#1337), we patch the resolved JSON in place rather than
    asking them to clone the preset just to switch a plate. Returns the
    original string unchanged when the JSON can't be parsed or isn't a
    dict — the slicer will then run with whatever the preset originally
    specified, which is the safe fall-back path.
    """
    try:
        profile = json.loads(process_json)
    except json.JSONDecodeError:
        logger.warning("Bed-type override skipped: process profile is not valid JSON")
        return process_json
    if not isinstance(profile, dict):
        return process_json
    profile["curr_bed_type"] = bed_type
    return json.dumps(profile)


def _source_plate_colours(model_bytes: bytes) -> list[str]:
    """Per-slot colours the source 3MF was designed with, or ``[]``.

    Read from ``project_settings.config`` rather than ``slice_info.config``:
    the latter records the colour the file was *last sliced* with, which for a
    source that never carried one is the slicer's own #00AE42 default — the
    exact value #2977 is about, so using it as a fallback would be circular.
    STL and mesh-only 3MF sources have no project settings and yield ``[]``.
    """
    from io import BytesIO

    try:
        with zipfile.ZipFile(BytesIO(model_bytes), "r") as zf:
            return [str(f.get("color") or "") for f in extract_project_filaments_from_3mf(zf)]
    except (zipfile.BadZipFile, OSError, ValueError):
        return []


def _preset_default_colour(profile: dict) -> str:
    """A filament preset's own ``default_filament_colour``, or ``""``.

    OrcaSlicer's third-party vendor profiles carry this; Bambu Studio's
    bundled BBL filament profiles carry it nowhere (checked across the whole
    shipped `resources/profiles/BBL/filament/` tree — zero occurrences), which
    is why it can only ever be one link in the chain and never the whole fix.

    It is read here and rewritten as ``filament_colour`` because the CLI does
    not read it itself. Measured against a 02.08.02.61 sidecar: a profile
    carrying only ``default_filament_colour: ["#FF00FF"]`` still slices to
    ``filament_colour: ["#00AE42"]``. Bambu Studio consumes the default in the
    GUI when a project is created, not in ``--load-filaments``.
    """
    return first_slicer_string(profile.get("default_filament_colour")) or ""


def _patch_filament_colours(
    filament_jsons: list[str],
    requested: list[str],
    model_bytes: bytes,
) -> list[str]:
    """Write ``filament_colour`` onto each resolved filament profile (#2977).

    Neither slicer stores a colour on a filament *preset* — it is a per-project
    property their GUIs set from the plate — so a CLI slice with no colour
    supplied records Bambu Studio's compiled-in default for every slot. That
    default is `#00AE42`, which is why every internal-slicer output was green
    regardless of the filament picked, and why the print dialog's AMS mapping
    reported a colour mismatch against whatever was actually loaded.

    Per slot, first non-empty of:

    1. the caller's explicit colour (the SliceModal's per-slot swatch),
    2. the preset's own ``default_filament_colour``,
    3. the colour the source 3MF's plate was designed with.

    All three empty means the slot is left untouched rather than being given a
    guess: the slicer's default is then still wrong, but it is at least the
    same wrong value the file would have had before this function existed.

    Returns a new list; a profile that isn't parseable JSON is passed through
    unchanged, on the same reasoning as ``_patch_process_bed_type`` — a colour
    is not worth failing a slice that would otherwise succeed.
    """
    source_colours = _source_plate_colours(model_bytes) if filament_jsons else []
    patched: list[str] = []
    for i, raw in enumerate(filament_jsons):
        try:
            profile = json.loads(raw)
        except json.JSONDecodeError:
            logger.warning("Filament colour skipped for slot %d: profile is not valid JSON", i + 1)
            patched.append(raw)
            continue
        if not isinstance(profile, dict):
            patched.append(raw)
            continue
        colour = (
            (requested[i].strip() if i < len(requested) and requested[i] else "")
            or _preset_default_colour(profile)
            or (source_colours[i].strip() if i < len(source_colours) and source_colours[i] else "")
        )
        if not colour:
            patched.append(raw)
            continue
        # One-element array: the same shape the CLI uses for every other
        # per-filament field (`filament_type`, `filament_vendor`), and the
        # shape a `--load-filaments` profile is parsed as. A bare string is
        # accepted by the JSON parser but not by the config deserialiser.
        profile["filament_colour"] = [colour]
        patched.append(json.dumps(profile))
    return patched


# Support-related keys we lift from the source 3MF's project_settings.config
# into the picked process preset before `--load-settings` sees it (#1881).
# BambuStudio's shipped process presets ("0.20mm Standard @BBL H2D" etc.)
# define `enable_support: 0` as their default — supports are a per-print
# decision, not a per-quality one. `--load-settings` is authoritative, so
# without preserving these fields the source's per-project support intent
# (supports on, PVA in the interface slot, tree vs normal) gets discarded
# and the slicer produces a single-material output with no supports at all.
_SOURCE_PROCESS_SUPPORT_KEYS_TO_PRESERVE = (
    "enable_support",
    "support_filament",
    "support_interface_filament",
    "support_type",
)


def _declined_source_keys(offered: list[DesignOverride], requested: list[str] | None) -> set[str]:
    """Settings the file offered and the caller left unticked (#2942).

    The slice dialog lists what the designer changed and applies only the keys
    that are switched on, so the answer to "which of these does this slice
    want" is already in the request. This reads the other half of it — the
    ones that were on offer and turned down — which the support carry-over
    below must not put back.

    ``requested`` of ``None`` is a caller that predates the per-key choice and
    so cannot have declined anything; an empty list is one that was shown the
    file's settings and took none. Collapsing those two into "nothing
    selected" is what made an empty panel indistinguishable from an old
    client, and only one of them means the user said no.
    """
    if requested is None:
        return set()
    return {override.key for override in offered} - set(requested)


def _patch_process_support_settings(
    process_json: str,
    source_3mf_bytes: bytes,
    declined: set[str] | frozenset[str] = frozenset(),
) -> str:
    """Overlay the source 3MF's support configuration onto the process JSON.

    The carry is deliberately one-way: a source can switch supports *on*,
    never off (#2820). The original #1881 rule was "source wins in both
    directions", which quietly stripped supports from every custom process
    preset that enabled them — a MakerWorld download nearly always ships
    `enable_support: 0`, so the reporter's own preset (supports on, normal
    (auto)) came back out of the slicer disabled and set to tree(auto).
    Nothing is lost by not carrying the off direction: a process preset
    with supports *on* is by definition a deliberate user preset, since
    Bambu's shipped ones all ship them off.

    ``declined`` names keys the caller offered the user as the file's own
    (#2622) and that the user left unticked, which this carry must then not
    reinstate behind their back (#2942). It is empty for a source that offers
    nothing — an OrcaSlicer export carries no ``different_settings_to_system``,
    so there is nothing to tick and #1881's blanket carry still applies — and
    for a client that predates the per-key ticks.

    Only fires on 3MF sources — STL / STEP don't carry `project_settings.
    config`. Silently no-ops when the source doesn't have the config, has
    a malformed one, or when the process JSON isn't parseable — the slice
    then runs with the process preset's own defaults, which is the safe
    fall-back for both this bug and the pre-fix behaviour.
    """
    from io import BytesIO

    try:
        with zipfile.ZipFile(BytesIO(source_3mf_bytes), "r") as zf:
            if "Metadata/project_settings.config" not in zf.namelist():
                return process_json
            src_cfg = json.loads(zf.read("Metadata/project_settings.config").decode("utf-8"))
    except (zipfile.BadZipFile, json.JSONDecodeError, UnicodeDecodeError, OSError, KeyError):
        return process_json
    if not isinstance(src_cfg, dict):
        return process_json
    if not supports_enabled_in_config(src_cfg):
        return process_json

    try:
        process_cfg = json.loads(process_json)
    except json.JSONDecodeError:
        return process_json
    if not isinstance(process_cfg, dict):
        return process_json

    carried = {
        key: src_cfg[key] for key in _SOURCE_PROCESS_SUPPORT_KEYS_TO_PRESERVE if key in src_cfg and key not in declined
    }
    if not carried:
        return process_json
    process_cfg.update(carried)
    # Logged because this is the one layer of the process JSON the user
    # can't see coming: the slice modal shows the picked preset's values,
    # so a carried key silently disagrees with what was on screen.
    logger.info(
        "Carried support settings from the source 3MF onto the process preset: %s",
        dict(sorted(carried.items())),
    )

    return json.dumps(process_cfg)


# The sidecar prefixes the slicer CLI's own error_string with this when the
# slicer ran and rejected the job (model off the bed, incompatible filament
# temps, range validation) — as opposed to the CLI crashing before it could
# evaluate the job at all.
_SLICER_REJECTION_MARKER = "Slicing failed with error from slicer:"

# The CLI writes its real diagnostic to stdout/stderr on the `[error]` level.
# Format is `[<timestamp>] [error] run <NNNN>: <message>` (or sometimes without
# the `run NNNN:` prefix). The bracketed timestamp is optional; the `[error]`
# tag is what we anchor on. Used to recover the actual rejection reason for
# the `error_string: "The input preset file is invalid and can not be parsed."`
# case (#1851) — the CLI emits that generic placeholder for every -5 exit
# including real preset-compat rejections, and the per-incident specifics
# only live in the stdout dump.
_CLI_ERROR_LINE_RE = re.compile(r"\[error\]\s*(?:run\s+\d+:\s*)?(.+?)\s*$", re.MULTILINE)

# The placeholder error_string Bambu Studio writes to result.json for any
# `--load-settings` parse / compat rejection (-5 exit). When the sidecar
# surfaces this, the real reason lives in the stdout `[error]` line that we
# mine via _CLI_ERROR_LINE_RE.
_INPUT_PRESET_INVALID_PLACEHOLDER = "The input preset file is invalid and can not be parsed."


def _slicer_rejection_message(error_text: str) -> str | None:
    """Extract the slicer's own rejection reason from a sidecar error string,
    or ``None`` when the failure is not a slicer content rejection.

    A content rejection means ``--load-settings`` *was* applied — the slicer
    got far enough to evaluate the model against the chosen printer and say
    no. Retrying with the 3MF's embedded settings would then only "succeed"
    by silently reverting to the source file's original printer, masking the
    real problem; such failures must reach the user instead.

    When the sidecar's `error_string` is Bambu Studio's generic
    "The input preset file is invalid and can not be parsed." placeholder
    (#1851) — emitted for every -5 exit, including the actual preset-compat
    rejections whose real reason is logged to stdout as
    `[error] run NNNN: <diagnostic>` — prefer the stdout `[error]` line so
    the user sees which preset clashed with which printer.
    """
    if _SLICER_REJECTION_MARKER not in error_text:
        return None
    reason = error_text.split(_SLICER_REJECTION_MARKER, 1)[1]
    # Mine the stdout/stderr dump for a more specific CLI diagnostic before
    # we trim it off below. Done first so the lookup window covers the full
    # response, not just the headline.
    cli_diagnostic_match = _CLI_ERROR_LINE_RE.search(reason)
    cli_diagnostic = cli_diagnostic_match.group(1).strip() if cli_diagnostic_match else None
    # Trim the sidecar's trailing exit-code note and any stderr/stdout dump.
    for cut in (": Slicer process failed", "\nstderr:", "\nstdout:"):
        idx = reason.find(cut)
        if idx != -1:
            reason = reason[:idx]
    reason = reason.strip() or None
    # When the headline is Bambu Studio's catch-all placeholder, the real
    # reason is in the stdout `[error]` line. Substitute it. The placeholder
    # by itself tells the user nothing about why their slice was rejected.
    if cli_diagnostic and (reason is None or reason == _INPUT_PRESET_INVALID_PLACEHOLDER):
        return cli_diagnostic
    return reason


async def _run_slicer_with_fallback(
    db: AsyncSession,
    *,
    model_bytes: bytes,
    model_filename: str,
    request: SliceRequest,
    current_user_id: int | None = None,
    job_id: int | None = None,
):
    """Validate presets, dispatch to the right sidecar, run the slicer with
    the auto-fallback for 3MF inputs whose `--load-settings` path crashes the
    CLI. Returns ``(SliceResult, used_embedded_settings: bool)``. Raises
    ``HTTPException`` for any caller-facing error.

    `current_user_id` is needed to resolve **cloud** presets — the cloud token
    is per-user when auth is enabled. For the legacy / local-only path it can
    be left ``None``.

    `job_id`: when set, a request_id is generated and a parallel poller
    pushes the sidecar's --pipe-fed progress events onto
    ``slice_dispatch.set_progress(job_id, ...)`` so the UI's persistent
    toast can show "Generating G-code (75%)" instead of just elapsed
    time. Pass None for synchronous routes that aren't tracked by the
    dispatcher.
    """
    from backend.app.api.routes.settings import get_setting
    from backend.app.services.preset_resolver import resolve_preset_ref
    from backend.app.services.slicer_api import (
        SlicerApiServerError,
        SlicerApiService,
        SlicerApiUnavailableError,
        SlicerInputError,
        SlicerTimeoutError,
        get_stall_timeout_seconds,
    )

    user: User | None = None
    presets: dict[str, str] = {}
    filament_jsons: list[str] = []
    # Resolve each slot via the source-aware resolver. The schema
    # validator has already normalised legacy `*_preset_id: int`
    # fields into `PresetRef(source='local', id=str(int))`, so all
    # three are guaranteed non-None here.
    if current_user_id is not None:
        user = await db.get(User, current_user_id)

    refs = {
        "printer": request.printer_preset,
        "process": request.process_preset,
    }
    for slot, ref in refs.items():
        assert ref is not None, "schema validator guarantees PresetRef is set"
        presets[slot] = await resolve_preset_ref(db, user, ref, slot)
    # Multi-color: resolve each filament slot in plate order. The schema
    # validator backfilled `filament_presets` from the legacy `filament_preset`
    # field for single-color callers, so this list is always non-empty.
    for ref in request.filament_presets:
        assert ref is not None, "schema validator guarantees filament list is non-None"
        filament_jsons.append(await resolve_preset_ref(db, user, ref, "filament"))

    # Give every slot a colour before anything else touches the list, so the
    # unused-slot substitution below propagates a complete profile rather than
    # one that still has to be patched afterwards (#2977).
    filament_jsons = _patch_filament_colours(filament_jsons, request.filament_colours, model_bytes)

    # Bed-type override (#1337): patch curr_bed_type onto the resolved
    # process JSON so the slicer's StaticPrintConfig pass picks up the
    # user's pick instead of whatever the process preset defaults to.
    # Without this, slicing an STL of ABS onto a process preset whose
    # default is "Cool Plate" fails with "Plate 1: Cool Plate does not
    # support filament 1" — the reporter's exact scenario.
    if request.bed_type:
        presets["process"] = _patch_process_bed_type(presets["process"], request.bed_type)

    # Slicer routing — pick the sidecar URL by preferred_slicer.
    # The per-install URL setting (Settings UI → Slicer card) wins; an
    # empty value falls back to the SLICER_API_URL / BAMBU_STUDIO_API_URL
    # env defaults defined in core/config.py.
    preferred = (await get_setting(db, "preferred_slicer")) or "bambu_studio"
    if preferred == "orcaslicer":
        configured = await get_setting(db, "orcaslicer_api_url")
        api_url = (configured or app_settings.slicer_api_url).strip()
    elif preferred == "bambu_studio":
        configured = await get_setting(db, "bambu_studio_api_url")
        api_url = (configured or app_settings.bambu_studio_api_url).strip()
    else:
        raise HTTPException(
            status_code=400,
            detail=f"Unknown preferred_slicer setting: '{preferred}'. Expected 'orcaslicer' or 'bambu_studio'.",
        )

    # Note: an earlier version of this code stripped Metadata/project_settings.
    # config + model_settings.config + slice_info.config + cut_information.xml
    # before forwarding the 3MF, the theory being that --load-settings would
    # then take precedence cleanly. That theory was wrong: model_settings.
    # config carries the plate definitions the CLI needs to map `--slice N`
    # to a real plate, and slice_info / project_settings supply baseline
    # config the CLI's StaticPrintConfig pass needs at all. Stripping ANY
    # of them caused the CLI to silently exit immediately after
    # "Initializing StaticPrintConfigs" — exit code 0, no result.json, no
    # stderr — which Node's child_process treated as failure and Bambuddy
    # then masked by falling back to slice_without_profiles using the
    # un-stripped bytes (and the source's embedded printer). Net effect:
    # every 3MF slice with profiles silently produced wrong-printer output.
    # Forwarding the original bytes lets --load-settings override the
    # specific fields the user changed (printer/process/filament) while
    # the embedded plate / model definitions remain intact.
    is_3mf = model_filename.lower().endswith(".3mf")
    primary_bytes = model_bytes
    if is_3mf:
        # Strip inherit/unset sentinels from Metadata/project_settings.config
        # so the CLI's StaticPrintConfig range validator accepts the file
        # (#1201, #3030). Surgical — keeps the config present, just removes
        # the offending keys; the supplied --load-settings (and the fallback's
        # embedded values for keys we didn't touch) still drive the slice.
        # The preview-slice path applies the same sanitiser in
        # ``slice_preview.get_preview_filaments``.
        primary_bytes = sanitize_project_settings_sentinels(primary_bytes)

        # #2622: the process settings the file's designer moved off the stock
        # preset. Read once — the support patch below needs to know which of
        # them the user was shown, and the carry after it needs their values.
        design_offered = extract_design_process_overrides(primary_bytes)

        declined_from_file = _declined_source_keys(design_offered, request.design_overrides)

        # #1881: preserve the source 3MF's support configuration on top of
        # the picked process preset. Bambu's shipped process presets set
        # `enable_support: 0` by default (supports are a per-print, not
        # per-quality, decision); `--load-settings` is authoritative so
        # without patching, the source's `enable_support: 1` + support-slot
        # assignments get discarded and the slice comes out single-material
        # with a PVA slot loaded but never used. Bounded by the ticks: this
        # runs for a source that offers no per-key choice at all, and for the
        # keys of one that does but whose ticks the user left on.
        presets["process"] = _patch_process_support_settings(
            presets["process"], primary_bytes, declined=declined_from_file
        )

        # Carry the designer's tweaks onto the picked preset. BambuStudio
        # records exactly which keys deviate from the system preset in
        # `different_settings_to_system`, so a MakerWorld author's 5 walls /
        # 100% infill / 0.1mm first layer survive a re-slice for another printer
        # instead of being flattened by --load-settings. Opt-in per key: only the
        # keys the caller names are applied, and only if the source really lists
        # them as changed. Runs after the #1881 support patch so an explicit
        # design pick wins over the blanket support carry-over.
        if request.design_overrides:
            presets["process"] = apply_design_overrides(
                presets["process"],
                design_offered,
                request.design_overrides,
            )

    # The user's own edits from the slice modal's settings panel. Applied last
    # and for every model type (not just 3MF): unlike the two patches above this
    # doesn't read anything out of the source file, it is what the user typed.
    # Last write wins, so an explicit choice beats both the carried support
    # config (#1881) and the designer's tweaks (#2622).
    if request.process_overrides:
        presets["process"] = apply_process_overrides(presets["process"], request.process_overrides)

    used_embedded_settings = False
    # "Slice as designed" (#2611): honour the file's embedded
    # project_settings.config instead of the picked profile triplet. Only
    # meaningful for a 3MF that actually carries embedded settings; the UI
    # gates the toggle on the picked printer matching the design's target,
    # so this path never re-targets across printer models.
    embedded_mode = bool(request.use_embedded_settings and is_3mf)
    # The schema takes either slicer's plate names, but Bambu Studio slices a
    # name it does not know as Cool Plate (#3359).
    if not embedded_mode and preferred == "bambu_studio" and request.bed_type in ORCA_ONLY_BED_TYPES:
        raise HTTPException(
            status_code=400,
            detail=f"Bambu Studio has no {request.bed_type}. Pick another build plate or slice with OrcaSlicer.",
        )
    # Bounds silence rather than total slicing time (#2730), so a heavy model
    # that keeps reporting progress runs to completion however long it takes.
    service = SlicerApiService(api_url, timeout_seconds=await get_stall_timeout_seconds(db))

    # #1493: cross-nozzle-class re-slice (single <-> dual). Without
    # intervention the slicer rejects with either "G-code in unprintable
    # area of multi-extruder printers" (the source's X1C-coordinate layout
    # lands in the H2D's per-nozzle dead zone) or — worse — segfaults
    # inside ZFiller's polygon clipping when the geometry pipeline trips
    # on the cross-class transition. Forwarding the sidecar's --arrange
    # flag for these cases lets BambuStudio reposition objects for the
    # target bed and reconcile the embedded project_settings.config
    # against the new printer, the same way the GUI's "Switch Printer"
    # operation does. --arrange WILL reposition objects, so we only
    # enable it on a true class crossing — same-printer slices keep the
    # user's deliberate layout. The bed-type and arrange flags are
    # orthogonal so this decision doesn't interact with the #1337 build-
    # plate override.
    cross_class_arrange = False
    if is_3mf:
        from backend.app.services.slicer_3mf_convert import (
            extract_source_printer_model,
        )
        from backend.app.utils.printer_models import is_dual_nozzle_model

        source_model = extract_source_printer_model(primary_bytes)
        target_model = await _resolve_target_printer_model(db, user, request)
        if source_model and target_model and is_dual_nozzle_model(source_model) != is_dual_nozzle_model(target_model):
            logger.info(
                "Cross-nozzle-class re-slice (%s -> %s): enabling --arrange so BS reconciles "
                "the embedded project layout against the target printer",
                source_model,
                target_model,
            )
            cross_class_arrange = True

    # #2548: the user can also ask for either layout pass per-slice. Arrange
    # is a union with the cross-class decision above — a user opt-out must
    # not be able to switch off the flag that keeps a class-crossing slice
    # from crashing — while orient is user-driven only.
    arrange_flag = cross_class_arrange or request.auto_arrange
    orient_flag = request.auto_orient
    # When this slice is dispatcher-tracked, generate a request_id so
    # the sidecar publishes progress under it, and wire a callback that
    # forwards each frame onto SliceDispatchService.set_progress for the
    # status-poll endpoint to surface to the UI.
    progress_request_id: str | None = None
    progress_callback = None
    if job_id is not None:
        from uuid import uuid4

        from backend.app.services.slice_dispatch import slice_dispatch as _dispatch

        progress_request_id = str(uuid4())

        def _on_progress(snapshot: dict) -> None:
            _dispatch.set_progress(job_id, snapshot)

        progress_callback = _on_progress
    # SliceModal lets the user pick a filament profile per slot, but each
    # plate uses only a subset of the slots. The unused-slot dropdowns get
    # whatever default the modal serves up — and a heterogeneous default
    # (e.g. ABS in slot 2 next to a PLA in the used slot 1) makes
    # BambuStudio reject the slice with "the temperature difference of
    # the filaments used is too large" (exit 194) even though the G-code
    # never touches the unused slot; a default scoped to another printer
    # gets it rejected with "filament preset (slot N) is not compatible
    # with printer …" (#2628). Replace unused-slot entries with the
    # plate's lowest used slot before the real slice so the loaded set is
    # materially homogeneous and printer-correct.
    #
    # ``plate`` is absent for single-plate and STL sources — the SliceModal
    # skips the picker and omits the field — and absent means plate 1, the
    # same reading as ``plate_num`` further down and as the schema's own
    # description. Treating it as "unknown plate" instead is what left every
    # single-plate 3MF unsubstituted (#2711): a MakerWorld project defining
    # four filaments but painting only one reached the CLI with the other
    # three still holding presets baked into the source for a different
    # printer, and the slice died on the first of them.
    #
    # ``plate=0`` is the slice-all sentinel, not a plate: every slot is used
    # by some plate, so there is nothing to substitute. It has to be excluded
    # explicitly because the support-filament slots unioned in below are
    # read from the project config and are not plate-scoped — they would
    # survive the (empty) geometry lookup for plate 0 and become the anchor,
    # collapsing every colour of a slice-all onto the support filament.
    if is_3mf and request.plate != 0:
        from backend.app.services.slicer_3mf_convert import substitute_unused_plate_filaments

        filament_jsons = substitute_unused_plate_filaments(primary_bytes, request.plate or 1, filament_jsons)

    # Arrange slice-all loop (#1493): when the user asks for ``plate=0``
    # (all plates) AND arrange is on, ``--slice 0 --arrange 1``
    # consolidates every plate's objects onto a single target bed (BS's
    # ``--arrange`` is project-wide) — either packing them all together or
    # rejecting with "Some objects are located over the boundary of the
    # heated bed" when nothing fits. Slice each plate independently with
    # ``--arrange 1`` and merge the per-plate outputs into one multi-plate
    # 3MF instead. Slice-all without arrange goes through the regular path
    # below — the sidecar's native ``--slice 0`` produces the right shape
    # directly.
    #
    # Keyed on ``arrange_flag``, not just the cross-class decision: the
    # project-wide collapse is a property of ``--arrange`` itself, so a
    # user-requested arrange over all plates (#2548) hits it identically.
    # Orient doesn't — it rotates objects where they stand and never moves
    # one between plates — so it isn't part of this condition.
    use_arrange_slice_all = arrange_flag and request.plate == 0 and request.export_3mf

    try:
        try:
            if use_arrange_slice_all:
                from backend.app.services.slicer_3mf_convert import (
                    count_plates_in_3mf,
                    merge_plate_3mfs,
                )

                plate_count = count_plates_in_3mf(primary_bytes)
                if plate_count == 0:
                    raise HTTPException(
                        status_code=400,
                        detail=(
                            "Couldn't read plate count from the source 3MF for cross-class "
                            "slice-all. The source may be malformed or missing "
                            "Metadata/model_settings.config."
                        ),
                    )
                logger.info(
                    "Arrange slice-all: looping over %d plates with --arrange per plate, then merging "
                    "(embedded_settings=%s)",
                    plate_count,
                    embedded_mode,
                )
                from backend.app.services.slicer_api import SliceResult

                per_plate_results: list[tuple[int, SliceResult]] = []

                # Forward the same progress request_id + callback to each
                # per-plate sub-call so the toast keeps showing the
                # sidecar's stage messages ("Generating G-code 45%…").
                # The sub-calls run sequentially, so the poller for plate
                # N is cancelled before plate N+1's poller starts — no
                # cross-talk between plate streams. Wrap the callback to
                # surface "(plate N/M)" alongside the slicer's stage
                # message so the user sees progress through the whole
                # multi-plate loop, not just one plate at a time.
                def _wrap_progress_for_plate(plate_num: int, total: int):
                    if progress_callback is None:
                        return None

                    def _cb(snapshot: dict) -> None:
                        snapshot = dict(snapshot)
                        snapshot["multi_plate_index"] = plate_num
                        snapshot["multi_plate_count"] = total
                        progress_callback(snapshot)

                    return _cb

                for plate_num in range(1, plate_count + 1):
                    plate_cb = _wrap_progress_for_plate(plate_num, plate_count)
                    # "Slice as designed" has to take the loop too, not skip
                    # it: the project-wide collapse is caused by --arrange,
                    # and which config drives the slice has no bearing on
                    # that. Same call, minus --load-settings.
                    if embedded_mode:
                        per_plate = await service.slice_without_profiles(
                            model_bytes=primary_bytes,
                            model_filename=model_filename,
                            plate=plate_num,
                            export_3mf=True,
                            arrange=True,
                            orient=orient_flag,
                            request_id=progress_request_id,
                            on_progress=plate_cb,
                        )
                    else:
                        per_plate = await service.slice_with_profiles(
                            model_bytes=primary_bytes,
                            model_filename=model_filename,
                            printer_profile_json=presets["printer"],
                            process_profile_json=presets["process"],
                            filament_profile_jsons=filament_jsons,
                            plate=plate_num,
                            export_3mf=True,
                            arrange=True,
                            orient=orient_flag,
                            request_id=progress_request_id,
                            on_progress=plate_cb,
                        )
                    per_plate_results.append((plate_num, per_plate))

                # Merge the N single-plate 3MFs into one multi-plate 3MF.
                # ``primary_bytes`` is the source 3MF: it carries the
                # original per-plate previews the slicer's --arrange
                # pass doesn't regenerate, so the merger can fall back
                # to those for each plate's cover image.
                merged_bytes = merge_plate_3mfs(
                    [(n, r.content) for n, r in per_plate_results],
                    source_3mf_bytes=primary_bytes,
                )
                # Synthetic SliceResult: totals are the sum of each
                # plate's so the archive card shows the project's print
                # time and filament use, not just plate 1's.
                result = SliceResult(
                    content=merged_bytes,
                    print_time_seconds=sum(r.print_time_seconds for _, r in per_plate_results),
                    filament_used_g=sum(r.filament_used_g for _, r in per_plate_results),
                    filament_used_mm=sum(r.filament_used_mm for _, r in per_plate_results),
                )
                # Report the path honestly: the loop can run either way, and
                # the UI reads this flag to tell the user whose settings won.
                used_embedded_settings = embedded_mode
            elif embedded_mode:
                # No --load-settings: feed the CLI the file's own
                # project_settings.config untouched so the designer's tweaks
                # (walls, infill, etc.) drive the slice. primary_bytes is
                # already sentinel-sanitised above, the same bytes the
                # crash-fallback uses. The resolved presets go unused here.
                # Arrange / orient still apply: they are CLI actions on the
                # geometry, not settings the embedded config could carry.
                result = await service.slice_without_profiles(
                    model_bytes=primary_bytes,
                    model_filename=model_filename,
                    plate=request.plate,
                    export_3mf=request.export_3mf,
                    arrange=arrange_flag,
                    orient=orient_flag,
                    request_id=progress_request_id,
                    on_progress=progress_callback,
                )
                used_embedded_settings = True
            else:
                result = await service.slice_with_profiles(
                    model_bytes=primary_bytes,
                    model_filename=model_filename,
                    printer_profile_json=presets["printer"],
                    process_profile_json=presets["process"],
                    filament_profile_jsons=filament_jsons,
                    plate=request.plate,
                    export_3mf=request.export_3mf,
                    arrange=arrange_flag,
                    orient=orient_flag,
                    request_id=progress_request_id,
                    on_progress=progress_callback,
                )
        except SlicerApiServerError as exc:
            rejection = _slicer_rejection_message(str(exc))
            if rejection:
                # The slicer ran and rejected the job for a content reason —
                # the chosen printer/process/filament *were* applied. Falling
                # back to embedded settings would silently re-slice for the
                # source 3MF's original printer and hide the real problem
                # (e.g. re-slicing an H2D model for an X1C: the object is off
                # the smaller bed). Surface the slicer's reason instead.
                raise HTTPException(status_code=400, detail=rejection) from exc
            if not is_3mf or embedded_mode:
                # embedded_mode already sliced with the file's own settings —
                # there is nothing to fall back TO, so surface the server
                # error (the outer handler turns it into a 502) instead of
                # re-running the same embedded slice.
                raise
            if use_arrange_slice_all:
                # The fallback is a single ``--slice 0`` call, and with
                # arrange on that collapses every plate onto one bed — the
                # exact outcome the per-plate loop above exists to avoid.
                # Retrying would hand back a one-plate result for a job the
                # user asked to slice as N, which reads as a Bambuddy bug
                # rather than a slicer failure. Surface the error instead.
                raise
            logger.warning(
                "Slicer CLI failed on the --load-settings path for %s (%s); retrying with embedded settings",
                model_filename,
                exc,
            )
            # Forward the same request_id + callback so the toast's live
            # progress keeps updating across the fallback retry instead
            # of going blank for the rest of the slice. Use the sanitised
            # bytes — the embedded-settings path also reads the same
            # project_settings.config and the same range validator runs
            # there too, so without sanitisation the fallback would die
            # on the same sentinel error (#1201). The SliceModal flags
            # the difference to the user via used_embedded_settings.
            # Carry the layout flags across too — the retry is meant to
            # differ from the failed attempt only in where the print
            # config came from, so dropping them here would silently
            # produce an un-arranged result the user did ask for.
            result = await service.slice_without_profiles(
                model_bytes=primary_bytes,
                model_filename=model_filename,
                plate=request.plate,
                export_3mf=request.export_3mf,
                arrange=arrange_flag,
                orient=orient_flag,
                request_id=progress_request_id,
                on_progress=progress_callback,
            )
            used_embedded_settings = True
    except SlicerInputError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except SlicerTimeoutError as exc:
        # 504, not 502: the sidecar answered for the whole run, we stopped
        # waiting. Reported separately so the user is told the slice ran out of
        # time and where to change that, rather than that the sidecar is
        # unreachable — which is what a read timeout used to look like (#2730).
        raise HTTPException(status_code=504, detail=str(exc)) from exc
    except SlicerApiServerError as exc:
        raise HTTPException(status_code=502, detail=str(exc)) from exc
    except SlicerApiUnavailableError as exc:
        raise HTTPException(status_code=502, detail=str(exc)) from exc
    finally:
        await service.close()

    # Backstop for #2838. Only the standard tier, and only when the presets we
    # sent were actually used: there the sidecar resolved a bundled preset by
    # name and the bundle guarantees the start G-code, so its absence is a
    # sidecar defect we can name. A cloud, local or Orca-cloud preset carries
    # its own start G-code, and the embedded-settings fallback prints the
    # source file's — both are the user's to author, and refusing them here
    # would be us second-guessing a profile we did not resolve.
    if (
        not used_embedded_settings
        and request.printer_preset is not None
        and request.printer_preset.source == "standard"
        and start_gcode_is_missing(result.content, export_3mf=bool(request.export_3mf))
    ):
        logger.error(
            "Slice for printer preset %r came back without start G-code (%s); refusing it",
            request.printer_preset.id,
            "3mf" if request.export_3mf else "gcode",
        )
        raise HTTPException(status_code=502, detail=missing_start_gcode_message(request.printer_preset.id))

    # Found while investigating #2977: a filament preset the sidecar's bundle
    # cannot resolve is not an error there — the CLI inherits nothing and
    # slices with its own defaults, so a PETG pick comes back as PLA at 200 C.
    # Warned rather than refused: the file prints, and the user may well have
    # meant to slice with a profile their sidecar image predates. Skipped on
    # the embedded-settings path, which sends no filament profiles for the
    # bundle to resolve in the first place.
    if not used_embedded_settings:
        unresolved = unresolved_filament_slots(result.content, export_3mf=bool(request.export_3mf))
        if unresolved:
            logger.warning(
                "%s",
                unresolved_filament_message(unresolved, [ref.id for ref in request.filament_presets]),
            )

    return result, used_embedded_settings


def _canonical_printer_model(raw: str | None) -> str | None:
    """Normalise a printer-preset name / ``printer_model`` field to a canonical
    model code. Strips the BambuStudio ``"# "`` user-clone prefix and the
    ``" 0.4 nozzle"`` variant suffix that preset names carry but bare model
    names don't — without this, ``"Bambu Lab H2D 0.4 nozzle"`` wouldn't
    normalise to ``H2D``."""
    import re

    from backend.app.utils.printer_models import normalize_printer_model

    if not raw:
        return None
    cleaned = str(raw).strip()
    if cleaned.startswith("# "):
        cleaned = cleaned[2:].strip()
    cleaned = re.sub(r"\s+0\.\d+\s+nozzle$", "", cleaned, flags=re.IGNORECASE)
    return normalize_printer_model(cleaned) if cleaned else None


async def _resolve_target_printer_model(db: AsyncSession, user: User | None, request: SliceRequest) -> str | None:
    """Best-effort: the printer model a slice request targets.

    Returns ``None`` when it can't be determined (the nozzle-class guard
    then simply doesn't fire — fail-open, never blocks a slice spuriously).
    """
    from backend.app.services.preset_resolver import resolve_preset_ref

    if request.printer_preset is None:
        return None
    try:
        printer_json = await resolve_preset_ref(db, user, request.printer_preset, "printer")
        data = json.loads(printer_json)
        if not isinstance(data, dict):
            return None
        return _canonical_printer_model(
            data.get("printer_model") or data.get("printer_settings_id") or data.get("name")
        )
    except Exception:
        return None


async def guard_nozzle_class_reslice(
    db: AsyncSession, user: User | None, request: SliceRequest, source_model: str | None
) -> None:
    """No-op guard, retained for call-site compatibility.

    Cross-nozzle-class re-slicing is handled by ``_run_slicer_with_fallback``'s
    two-pass conversion (#1493): a 1mm cube is sliced with the target triplet
    via ``slice_with_profiles`` to produce a fresh target-shaped
    ``Metadata/project_settings.config``, which is then spliced into the
    source 3MF before the real slice. So this guard never needs to block
    anymore.

    The function and its call sites in ``archives.py`` / the library re-slice
    route are kept so external pinned-version forks and downstream patches
    don't break, but it does nothing on a successful slice path. If the
    two-pass conversion fails inside the slicer, the existing
    ``SlicerApiServerError`` / ``_slicer_rejection_message`` plumbing
    surfaces the CLI's actual error to the user — which is more informative
    than the old "isn't supported yet" 400 the guard used to raise.
    """
    return None


# slice_and_persist's default owner: the user it slices for.
_CALLER = object()


async def slice_and_persist(
    db: AsyncSession,
    *,
    model_bytes: bytes,
    model_filename: str,
    folder_id: int | None,
    extra_metadata: dict | None,
    request: SliceRequest,
    current_user_id: int | None,
    job_id: int | None = None,
    owner_id: int | None | object = _CALLER,
) -> SliceResponse:
    """Slice a model and save the result as a new ``LibraryFile`` in
    ``folder_id`` (same folder as the source by convention).

    ``current_user_id`` resolves cloud presets; the new file is ``owner_id``'s,
    by default the same user.

    Always exports as ``.gcode.3mf`` so the existing library thumbnail
    pipeline works on the new file. Plain ``.gcode`` would have no
    embedded thumbnail to extract.
    """
    from backend.app.services.archive import ThreeMFParser

    library_request = request.model_copy(update={"export_3mf": True})

    result, used_embedded_settings = await _run_slicer_with_fallback(
        db,
        model_bytes=model_bytes,
        model_filename=model_filename,
        request=library_request,
        current_user_id=current_user_id,
        job_id=job_id,
    )

    # Same reduction as the archive sink: ``model_filename`` may be built from
    # the source's embedded ``print_name``, which is free text (#2832). Managed
    # storage names the file after a UUID and never sees this, but an external
    # folder writes it verbatim, where a "/" would mean a directory nobody
    # created -- and the library row shows it either way.
    base_name = model_filename.rsplit(".", 1)[0]
    safe_base = safe_path_component(base_name, fallback="sliced", max_bytes=MAX_FILENAME_BYTES - len(b".gcode.3mf"))
    out_filename = f"{safe_base}.gcode.3mf"
    # Write next to the source when the source lives on an external mount
    # (#2810). The folder is loaded here rather than passed in because every
    # caller already has only the id.
    target_folder: LibraryFolder | None = None
    if folder_id is not None:
        folder_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
        target_folder = folder_result.scalar_one_or_none()
    out_path, out_is_external, external_fallback = _resolve_slice_destination(target_folder, out_filename)
    if out_is_external:
        # _unique_external_name may have suffixed it; the library row has to
        # show the name the file actually has on the share, or the two drift.
        out_filename = out_path.name
    if external_fallback:
        logger.warning(
            "Slice output for %s stored in managed library instead of external folder %s: %s",
            model_filename,
            target_folder.external_path if target_folder else None,
            external_fallback,
        )
    # BS/Orca CLIs skip plate_N.png in headless --export-3mf — render +
    # inject server-side so the library card has a thumbnail. Best-effort:
    # no-op when the slicer did embed thumbs (desktop Studio path), and
    # falls through to the unmodified bytes on any render error. In a thread:
    # a large plate renders for seconds, and on the event loop that stalled
    # every request and printer connection for as long (#3135).
    result = result._replace(content=await asyncio.to_thread(inject_plate_thumbnails_if_missing, result.content))
    out_path.write_bytes(result.content)

    # Extract thumbnail from the produced 3MF so the library card shows a
    # preview. Failures here aren't fatal — the file is still useful
    # without a thumbnail.
    thumbnail_relative: str | None = None
    parsed_metadata: dict = {}
    try:
        parser = ThreeMFParser(str(out_path))
        parsed = parser.parse()
        thumb_data = parsed.get("_thumbnail_data")
        thumb_ext = parsed.get("_thumbnail_ext", ".png")
        if thumb_data:
            thumb_filename = f"{uuid.uuid4().hex}{thumb_ext}"
            thumb_path = get_library_thumbnails_dir() / thumb_filename
            thumb_path.write_bytes(thumb_data)
            thumbnail_relative = to_relative_path(thumb_path)
        cleaned = _clean_3mf_metadata(parsed)
        if isinstance(cleaned, dict):
            parsed_metadata = cleaned
    except Exception as exc:
        logger.warning("Failed to parse sliced 3MF metadata for %s: %s", out_filename, exc)

    # Drop the embedded `print_name` (see _without_print_name) so the sliced
    # row's display falls back to its ".gcode.3mf" filename instead of the
    # source file's project title, which would make the two indistinguishable.
    metadata: dict = dict(_without_print_name(parsed_metadata) or {})
    # Some slicer-sidecar builds leave the X-Filament-Used-* response headers
    # unset, so result.filament_used_g/_mm arrive as 0 even for a real
    # multi-hour print. Fall back to the totals ThreeMFParser read from the
    # produced 3MF's own G-code header.
    filament_g = result.filament_used_g or parsed_metadata.get("filament_used_grams") or 0.0
    filament_mm = result.filament_used_mm or parsed_metadata.get("filament_used_mm") or 0.0
    metadata.update(
        {
            "print_time_seconds": result.print_time_seconds,
            "filament_used_g": filament_g,
            "filament_used_mm": filament_mm,
        }
    )
    if used_embedded_settings:
        metadata["used_embedded_settings"] = True
    if external_fallback:
        metadata["external_write_fallback"] = external_fallback
    if extra_metadata:
        metadata.update(extra_metadata)

    from backend.app.services.slice_dispatch import slice_dispatch as _dispatch

    # Nobody's, when the user was deleted while it sliced: the id may be someone else's now.
    file_owner_id = _dispatch.owner_now(job_id, current_user_id if owner_id is _CALLER else owner_id)  # type: ignore[arg-type]
    new_file = LibraryFile(
        folder_id=folder_id,
        is_external=out_is_external,
        filename=out_filename,
        file_path=_stored_file_path(out_path, out_is_external),
        # The on-disk payload is a ZIP container — the file_type must
        # record that so the preview endpoint opens it as a 3MF instead
        # of returning the ZIP bytes as text/plain (#1709 / yanglei1980).
        # Earlier code mis-typed sliced rows as "gcode" to share the
        # plain-G-code badge; that broke the embedded viewer. UI badges
        # and gates for "gcode.3mf" are explicit at the call sites.
        file_type="gcode.3mf",
        file_size=len(result.content),
        file_hash=hashlib.sha256(result.content).hexdigest(),
        thumbnail_path=thumbnail_relative,
        file_metadata=metadata,
        source_type="sliced",
        created_by_id=file_owner_id,
    )
    db.add(new_file)
    await db.commit()
    # No refresh: expire_on_commit=False keeps id/filename accessible, and
    # refreshing here flakes under pytest-xdist when teardown of a sibling
    # test races the SELECT.

    return SliceResponse(
        library_file_id=new_file.id,
        name=new_file.filename,
        print_time_seconds=result.print_time_seconds,
        filament_used_g=filament_g,
        filament_used_mm=filament_mm,
        used_embedded_settings=used_embedded_settings,
        external_write_fallback=external_fallback,
    )


async def slice_and_persist_as_archive(
    db: AsyncSession,
    *,
    model_bytes: bytes,
    model_filename: str,
    request: SliceRequest,
    source_archive,  # PrintArchive — hint kept loose to avoid cyclic import
    current_user_id: int | None,
    job_id: int | None = None,
    owner_id: int | None | object = _CALLER,
):
    """Slice a model and save the result as a new ``PrintArchive`` row,
    inheriting printer / project / makerworld metadata from the source
    archive. Always exports as a `.gcode.3mf` so the existing thumbnail
    and plates infrastructure (which expects a zip-shaped 3MF) works on
    the new archive. Returns ``SliceArchiveResponse``.
    """
    from backend.app.models.archive import PrintArchive
    from backend.app.schemas.slicer import SliceArchiveResponse
    from backend.app.services.archive import ThreeMFParser

    # Archive sinks always want a 3MF. The library route still respects the
    # caller's `export_3mf` flag; here we override.
    archive_request = request.model_copy(update={"export_3mf": True})

    result, used_embedded_settings = await _run_slicer_with_fallback(
        db,
        model_bytes=model_bytes,
        model_filename=model_filename,
        request=archive_request,
        job_id=job_id,
        current_user_id=current_user_id,
    )

    timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    printer_folder = str(source_archive.printer_id) if source_archive.printer_id is not None else "unassigned"

    # ``model_filename`` is built from the archive's display name, which comes
    # from the 3MF's own metadata and is whatever the model's author typed. A
    # "/" in it is a path separator, not a character: the joins below silently
    # gain a level and the write lands on a parent that was never created
    # (#2832). Reduce it to a single component first, leaving room for the
    # prefix and the extension wrapped around it.
    base_name = model_filename.rsplit(".", 1)[0]
    reserve = max(len(f"{timestamp}__sliced".encode()), len(b".gcode.3mf"))
    safe_base = safe_path_component(
        base_name, fallback=f"archive_{source_archive.id}", max_bytes=MAX_FILENAME_BYTES - reserve
    )
    out_filename = f"{safe_base}.gcode.3mf"
    archive_subdir = f"{timestamp}_{safe_base}_sliced"

    archive_dir = (
        app_settings.archive_dir / printer_folder / archive_subdir
    )  # SEC-PATH-OK: printer_folder = str(int|None); archive_subdir wraps safe_path_component output, asserted below
    out_path = archive_dir / out_filename  # SEC-PATH-OK: out_filename wraps safe_path_component output, asserted below
    # The sanitiser is what makes the two joins single-component; this is the
    # backstop that says so out loud, and would catch a future edit that reaches
    # around it. Checked before mkdir so a rejected path creates nothing.
    assert_under(app_settings.archive_dir, archive_dir, http=False)
    assert_under(app_settings.archive_dir, out_path, http=False)
    archive_dir.mkdir(parents=True, exist_ok=True)
    # See library-slice path: BS/Orca sidecar CLIs don't embed plate_N.png
    # in headless --export-3mf, so the produced 3MF often has no thumbnail
    # at all. Server-side render fills the gap; no-op when the slicer did
    # embed (desktop Studio path) and best-effort on any render error. Off the
    # event loop, like the library-slice path (#3135).
    result = result._replace(content=await asyncio.to_thread(inject_plate_thumbnails_if_missing, result.content))
    out_path.write_bytes(result.content)

    # Extract a thumbnail for the new archive card. Priority order:
    #   1. Source archive's ``Metadata/plate_{N}.png`` — the GUI-rendered
    #      preview of the same plate the user is re-slicing. Closer to
    #      "what's actually printing" than any other available image
    #      (with --arrange the layout may differ slightly, but objects
    #      and colours match).
    #   2. ``ThreeMFParser`` fallback chain on the sliced output: the
    #      slicer's own per-plate render if it wrote one, then the
    #      project-wide thumbnail under ``Auxiliaries/.thumbnails/``.
    # BambuStudio CLI frequently doesn't emit a fresh per-plate render
    # (slice writes the new gcode but leaves the preview slot empty),
    # so without (1) the card falls all the way through to the
    # MakerWorld-style cover art — visually unrelated to what the user
    # picked, see #1493 follow-up. Failures don't fail the slice — the
    # archive row is still useful without a thumbnail.
    plate_num = request.plate or 1
    thumbnail_path: str | None = None
    parsed_metadata: dict = {}

    src_3mf_path = app_settings.base_dir / source_archive.file_path
    source_plate_bytes = _read_3mf_entry(src_3mf_path, f"Metadata/plate_{plate_num}.png")
    if source_plate_bytes:
        thumb_dest = archive_dir / "thumbnail.png"
        thumb_dest.write_bytes(source_plate_bytes)
        thumbnail_path = str(thumb_dest.relative_to(app_settings.base_dir))

    try:
        parser = ThreeMFParser(str(out_path), plate_number=plate_num)
        parsed = parser.parse()
        if thumbnail_path is None:
            thumb_data = parsed.get("_thumbnail_data")
            thumb_ext = parsed.get("_thumbnail_ext", ".png")
            if thumb_data:
                thumb_dest = archive_dir / f"thumbnail{thumb_ext}"
                thumb_dest.write_bytes(thumb_data)
                thumbnail_path = str(thumb_dest.relative_to(app_settings.base_dir))
        parsed_metadata = {k: v for k, v in parsed.items() if not k.startswith("_")}
    except Exception as exc:
        logger.warning("Failed to parse sliced 3MF metadata for %s: %s", out_filename, exc)

    metadata = dict(source_archive.extra_data) if source_archive.extra_data else {}
    metadata.update(parsed_metadata)
    # Fall back to the produced 3MF's G-code-header totals when the sidecar
    # leaves the X-Filament-Used-* headers unset (result.filament_used_g == 0
    # even for a real multi-hour print).
    filament_g = result.filament_used_g or parsed_metadata.get("filament_used_grams") or 0.0
    filament_mm = result.filament_used_mm or parsed_metadata.get("filament_used_mm") or 0.0
    metadata.update(
        {
            "sliced_from_archive_id": source_archive.id,
            "print_time_seconds": result.print_time_seconds,
            "filament_used_g": filament_g,
            "filament_used_mm": filament_mm,
        }
    )
    if used_embedded_settings:
        metadata["used_embedded_settings"] = True

    # Prefer the actually-used filament list from the sliced output's
    # slice_info.config (parsed_metadata.filament_* — only entries with
    # used_g > 0). Falling back to the source_archive's list would
    # surface every project-wide AMS slot, including ones the picked
    # plate doesn't use (16+ swatches on the card for a 2-color print).
    new_filament_type = parsed_metadata.get("filament_type") or source_archive.filament_type
    new_filament_color = parsed_metadata.get("filament_color") or source_archive.filament_color

    # When the user re-slices for a different printer model than the source,
    # the source's printer_id (e.g. an H2D's "Workshop H2C") no longer
    # represents where the new archive can be reprinted. The archive card
    # and reprint modal both read printer_id first and only fall back to
    # sliced_for_model when it's None, so leaving the inherited id makes
    # the X1C-sliced card display the source H2D's printer name.
    # Same pitfall as the sliced_for_model copy a few lines below.
    new_target_model = parsed_metadata.get("sliced_for_model") or source_archive.sliced_for_model
    is_cross_model_reslice = (
        new_target_model is not None
        and source_archive.sliced_for_model is not None
        and new_target_model != source_archive.sliced_for_model
    )
    new_printer_id = None if is_cross_model_reslice else source_archive.printer_id

    from backend.app.services.slice_dispatch import slice_dispatch as _dispatch

    # Nobody's, when the user was deleted while it sliced: the id may be someone else's now.
    current_user_id = _dispatch.owner_now(job_id, current_user_id if owner_id is _CALLER else owner_id)  # type: ignore[arg-type]
    new_archive = PrintArchive(
        printer_id=new_printer_id,
        project_id=source_archive.project_id,
        filename=out_filename,
        file_path=str(out_path.relative_to(app_settings.base_dir)),
        file_size=len(result.content),
        content_hash=hashlib.sha256(result.content).hexdigest(),
        thumbnail_path=thumbnail_path,
        # Inherit identity from the source archive so the new entry shows
        # up alongside its sibling in the archives list.
        print_name=(source_archive.print_name or base_name) + " (re-sliced)",
        print_time_seconds=result.print_time_seconds,
        filament_used_grams=filament_g or None,
        filament_type=new_filament_type,
        filament_color=new_filament_color,
        layer_height=source_archive.layer_height,
        nozzle_diameter=source_archive.nozzle_diameter,
        # The re-sliced output is for whatever printer the user just picked,
        # not the source archive's printer — read the model the slicer baked
        # into the new 3MF, falling back to the source only if it's absent.
        # (Copying source_archive.sliced_for_model kept a cross-printer
        # re-slice, e.g. X1C→H2D, showing the old "X1C sliced" model.)
        sliced_for_model=parsed_metadata.get("sliced_for_model") or source_archive.sliced_for_model,
        # Build plate type that the sliced output was produced for (#1493
        # follow-up): the frontend's ArchiveCard reads ``archive.bed_type``
        # off the top-level column, not extra_data, so without this lift the
        # re-sliced card had no plate badge. ThreeMFParser pulls it from the
        # sliced 3MF's ``slice_info.config`` ``curr_bed_type``; if that's
        # absent (older sidecar / older slice profile) the source archive's
        # bed_type is the right default.
        bed_type=parsed_metadata.get("bed_type") or source_archive.bed_type,
        makerworld_url=source_archive.makerworld_url,
        designer=source_archive.designer,
        # Sliced, not printed: "archived", as archive_print files an unprinted
        # 3MF. It is listed under "Not printed" and is not counted as a
        # completed print; no started/completed_at either.
        status="archived",
        # Priced like any archive with grams, for the card's estimate.
        # Statistics read the print log, so nothing is counted that did not print.
        cost=await estimate_filament_cost(db, filament_g or None, new_filament_type),
        extra_data=metadata,
        created_by_id=current_user_id,
    )
    from backend.app.services.print_cost_estimate import refresh_plate_estimates

    refresh_plate_estimates(new_archive)
    db.add(new_archive)
    await db.commit()
    await db.refresh(new_archive)

    return SliceArchiveResponse(
        archive_id=new_archive.id,
        name=new_archive.print_name or out_filename,
        print_time_seconds=result.print_time_seconds,
        filament_used_g=filament_g,
        filament_used_mm=filament_mm,
        used_embedded_settings=used_embedded_settings,
    )


@router.post("/files/{file_id}/slice", status_code=202)
async def slice_library_file(
    file_id: int,
    request: SliceRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
    api_key_cloud_owner: User | None = Depends(resolve_api_key_cloud_owner),
    actor: User | ApiKeyActor | None = RequestActor,
):
    """Enqueue a slice job for a library file. Returns 202 + job_id; the
    slice runs in the background, the caller polls `GET /slice-jobs/{id}`.
    """
    from backend.app.core.database import async_session
    from backend.app.services.slice_dispatch import (
        http_exception_to_job_error,
        slice_dispatch,
    )

    src_result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    lib_file = src_result.scalar_one_or_none()
    # Per-row ownership gate. LIBRARY_UPLOAD alone let a READ_OWN caller (e.g. the
    # built-in Operators group) slice another user's model by raw id even though
    # GET on that id returned 404 — the sliced output was then attributed to and
    # downloadable by the requester. Enforce the same visibility the read routes
    # use before reading the source off disk. An API key is checked as its
    # owner (RequestActor); only auth off keeps can_read_all=True.
    can_read_all = actor is None or actor.has_permission(Permission.LIBRARY_READ_ALL.value)
    lib_file = _ensure_library_file_visible(lib_file, actor, can_read_all)

    src_lower = (lib_file.filename or "").lower()
    if src_lower.endswith(".step") or src_lower.endswith(".stp"):
        # Neither slicer's CLI can load STEP: OrcaSlicer 2.4.2 and BambuStudio
        # 02.07.01.62 both answer "Unknown file format. Input file must have
        # .stl, .obj, .amf(.xml) extension." Accepting the job here meant
        # reading the file, converting it and uploading it before the sidecar
        # rejected it as unparseable -- which reads as a corrupt model rather
        # than an unsupported format. Say so before any of that happens.
        raise HTTPException(
            status_code=400,
            detail=(
                "STEP files cannot be sliced. The OrcaSlicer and Bambu Studio command-line "
                "slicers load only STL and 3MF -- open the STEP in your slicer and export it "
                "as one of those first."
            ),
        )
    if not (src_lower.endswith(".stl") or src_lower.endswith(".3mf")):
        raise HTTPException(status_code=400, detail="Source file must be STL or 3MF")

    src_path = Path(app_settings.base_dir) / lib_file.file_path
    if not src_path.exists():
        raise HTTPException(status_code=404, detail="Source file missing on disk")

    # Capture inputs the bg task needs — the request DB session is closed
    # before the background task runs.
    model_bytes = src_path.read_bytes()
    folder_id = lib_file.folder_id
    source_lib_file_id = lib_file.id
    # API-keyed callers get None from the auth gate (auth.py keeps that
    # behaviour to avoid a wider scope expansion). Fall back to the API
    # key's owner so cloud-preset resolution can read the stored
    # cloud_token (#1182 follow-up).
    cloud_token_user = current_user or api_key_cloud_owner
    user_id = cloud_token_user.id if cloud_token_user else None
    # The sliced file is its owner's also when no one resolves cloud presets,
    # as a pipeline run's is.
    owner_id = user_id if user_id is not None else (actor.id if isinstance(actor, ApiKeyActor) else None)

    # If the source has a `print_name` in its metadata (BambuStudio always
    # sets this; OrcaSlicer often leaves it blank), derive the sliced
    # output's filename from it instead of the raw filename. The source
    # row's display already prefers print_name, so the sliced row's
    # filename ("Piggo the piggy bank.gcode.3mf") will match the source's
    # display name ("Piggo the piggy bank") with the gcode extension added.
    src_print_name = None
    if lib_file.file_metadata:
        candidate = lib_file.file_metadata.get("print_name")
        if isinstance(candidate, str) and candidate.strip():
            src_print_name = candidate.strip()
    src_ext = Path(lib_file.filename).suffix.lower() or ".3mf"
    model_filename = f"{src_print_name}{src_ext}" if src_print_name else lib_file.filename

    # Block a cross-nozzle-class re-slice (single-nozzle <-> H2D) up front.
    # Fires only when the source is itself a sliced file (carries
    # sliced_for_model); a plain un-sliced model has no source nozzle class.
    await guard_nozzle_class_reslice(
        db,
        cloud_token_user,
        request,
        (lib_file.file_metadata or {}).get("sliced_for_model"),
    )

    async def _run(job_id: int):
        async with async_session() as task_db:
            try:
                response = await slice_and_persist(
                    task_db,
                    model_bytes=model_bytes,
                    model_filename=model_filename,
                    folder_id=folder_id,
                    extra_metadata={"sliced_from_library_file_id": source_lib_file_id},
                    request=request,
                    current_user_id=user_id,
                    job_id=job_id,
                    owner_id=owner_id,
                )
            except HTTPException as exc:
                raise http_exception_to_job_error(exc) from exc
        return response.model_dump()

    job = await slice_dispatch.enqueue(
        kind="library_file",
        source_id=lib_file.id,
        source_name=lib_file.filename,
        owner_id=owner_id,
        run=_run,
    )
    return {
        "job_id": job.id,
        "status": job.status,
        "status_url": f"/api/v1/slice-jobs/{job.id}",
    }


@router.post("/files/{file_id}/print")
async def print_library_file(
    file_id: int,
    printer_id: int,
    # SECURITY.md SEC-AUTH-1: every route either has an explicit auth dep or
    # is in the route-auth-coverage allowlist. Gating the deprecation stub on
    # QUEUE_CREATE matches the replacement route (POST /queue/) and means
    # anonymous callers bounce at auth instead of seeing the deprecation
    # message.
    _: User | None = Depends(require_permission_if_auth_enabled(Permission.QUEUE_CREATE)),
):
    """Legacy direct library print endpoint. Use POST /queue/ instead."""
    logger.warning(
        "Gone API used: POST /library/files/%s/print?printer_id=%s; use POST /queue/ instead",
        file_id,
        printer_id,
    )
    raise HTTPException(
        status_code=410,
        detail="Direct library-file print has been removed. Create a print queue item with POST /queue/.",
    )


# ============ File Detail Endpoints ============


@router.get("/files/{file_id}", response_model=FileResponseSchema)
async def get_file(
    file_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Get a file by ID with full details."""
    user, can_read_all = auth_result
    result = await db.execute(
        LibraryFile.active().options(selectinload(LibraryFile.created_by)).where(LibraryFile.id == file_id)
    )
    file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)

    # Get folder name
    folder_name = None
    if file.folder_id:
        folder_result = await db.execute(select(LibraryFolder.name).where(LibraryFolder.id == file.folder_id))
        folder_name = folder_result.scalar()

    # Get project name
    project_name = None
    if file.project_id:
        project_result = await db.execute(select(Project.name).where(Project.id == file.project_id))
        project_name = project_result.scalar()

    # Get duplicates
    duplicates = []
    duplicate_count = 0
    if file.file_hash:
        dup_result = await db.execute(
            select(LibraryFile, LibraryFolder.name)
            .outerjoin(LibraryFolder, LibraryFile.folder_id == LibraryFolder.id)
            .where(
                LibraryFile.file_hash == file.file_hash,
                LibraryFile.id != file.id,
                LibraryFile.deleted_at.is_(None),
            )
        )
        for dup_file, dup_folder_name in dup_result.all():
            duplicates.append(
                FileDuplicate(
                    id=dup_file.id,
                    filename=dup_file.filename,
                    folder_id=dup_file.folder_id,
                    folder_name=dup_folder_name,
                    created_at=dup_file.created_at,
                )
            )
        duplicate_count = len(duplicates)

    # Extract key metadata fields
    print_name = None
    print_time = None
    filament_grams = None
    sliced_for_model = None
    if file.file_metadata:
        print_name = file.file_metadata.get("print_name")
        print_time = file.file_metadata.get("print_time_seconds")
        filament_grams = file.file_metadata.get("filament_used_grams")
        sliced_for_model = file.file_metadata.get("sliced_for_model")

    return FileResponseSchema(
        id=file.id,
        folder_id=file.folder_id,
        folder_name=folder_name,
        project_id=file.project_id,
        project_name=project_name,
        filename=file.filename,
        file_path=file.file_path,
        file_type=file.file_type,
        file_size=file.file_size,
        file_hash=file.file_hash,
        thumbnail_path=file.thumbnail_path,
        metadata=file.file_metadata,
        print_count=file.print_count,
        last_printed_at=file.last_printed_at,
        notes=file.notes,
        external_url=file.external_url,
        photos=list(file.photos or []),
        source_url=file.source_url,
        duplicates=duplicates if duplicates else None,
        duplicate_count=duplicate_count,
        created_by_id=file.created_by_id,
        created_by_username=file.created_by.username if file.created_by else None,
        created_at=file.created_at,
        updated_at=file.updated_at,
        print_name=print_name,
        print_time_seconds=print_time,
        filament_used_grams=filament_grams,
        sliced_for_model=sliced_for_model,
    )


@router.put("/files/{file_id}", response_model=FileResponseSchema)
async def update_file(
    file_id: int,
    data: FileUpdate,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_UPDATE_ALL,
            Permission.LIBRARY_UPDATE_OWN,
        )
    ),
):
    """Update a file's metadata."""
    user, can_modify_all = auth_result

    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    file = result.scalar_one_or_none()

    if not file:
        raise HTTPException(status_code=404, detail="File not found")

    # Ownership check
    if not can_modify_all:
        if file.created_by_id != user.id:
            raise HTTPException(status_code=403, detail="You can only update your own files")

    if data.filename is not None:
        # Bambu printer SD cards are FAT32/exFAT; reject the same set Bambu
        # Studio refuses on save so we fail here with a clear message
        # instead of an obscure FTP 553 at print time (#1540).
        try:
            validate_print_filename(data.filename)
        except InvalidFilenameError as e:
            raise HTTPException(status_code=400, detail=str(e)) from e
        file.filename = data.filename
        # No print_name to keep in sync — library files display by filename,
        # and _without_print_name strips the embedded 3MF Title on import (#1489).

    if data.folder_id is not None:
        target_folder = None
        if data.folder_id != 0:
            # library:update_all moves into any folder, as before (#3201).
            target_folder = await get_writable_folder(db, data.folder_id, None if can_modify_all else user)
        # Into or out of an external folder the bytes have to move too, and a
        # read-only one refuses them: that is the move route's work. Here only
        # folder_id changed, so the file showed in a folder its bytes were not
        # in (F8).
        unchanged = (target_folder.id if target_folder is not None else None) == file.folder_id
        if not unchanged and (file.is_external or (target_folder is not None and target_folder.is_external)):
            raise HTTPException(
                status_code=400,
                detail="Move a file into or out of an external folder with POST /api/v1/library/files/move",
            )
        file.folder_id = target_folder.id if target_folder is not None else None

    if data.project_id is not None:
        if data.project_id == 0:
            file.project_id = None
        else:
            # Verify project exists
            project_result = await db.execute(select(Project).where(Project.id == data.project_id))
            if not project_result.scalar_one_or_none():
                raise HTTPException(status_code=404, detail="Project not found")
            file.project_id = data.project_id

    if data.notes is not None:
        file.notes = data.notes if data.notes else None

    if data.external_url is not None:
        file.external_url = data.external_url.strip() or None

    await db.commit()
    await db.refresh(file)

    # Return full response. Bypass get_file's ownership gate — caller already
    # passed update_file's ownership gate above, so we re-fetch + serialise
    # directly instead of calling the route function (which would try to
    # evaluate its own Depends() at call time and trip a TypeError).
    return await get_file(file_id, db, auth_result=(None, True))


@router.delete("/files/{file_id}")
async def delete_file(
    file_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_DELETE_ALL,
            Permission.LIBRARY_DELETE_OWN,
        )
    ),
):
    """Move a file to the trash (soft-delete).

    The file's bytes and thumbnail stay on disk until the trash sweeper
    hard-deletes the row after the retention window (see #1008). External
    files skip the trash entirely — they can't be restored from disk and the
    underlying file is outside Bambuddy's control, so we just drop the DB
    record and thumbnail.
    """
    user, can_modify_all = auth_result

    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    file = result.scalar_one_or_none()

    if not file:
        raise HTTPException(status_code=404, detail="File not found")

    # Ownership check
    if not can_modify_all:
        if file.created_by_id != user.id:
            raise HTTPException(status_code=403, detail="You can only delete your own files")

    if file.is_external:
        # External files bypass the trash — just drop the DB row + our thumbnail.
        abs_thumb_path = to_absolute_path(file.thumbnail_path)
        if abs_thumb_path and abs_thumb_path.exists():
            try:
                abs_thumb_path.unlink()
            except OSError as e:
                logger.warning("Failed to delete thumbnail from disk: %s", e)
        from backend.app.services.library_trash import delete_dependent_variants, release_queue_references

        await delete_dependent_variants(db, [file.id])
        await release_queue_references(db, [file.id])
        remove_library_photos_dir(file.id)
        await db.delete(file)
        await db.commit()
        return {"status": "success", "message": "File deleted", "trashed": False}

    # Managed file: soft-delete. Sweeper removes bytes + thumbnail after retention.
    file.deleted_at = datetime.now(timezone.utc)
    await db.commit()
    return {"status": "success", "message": "File moved to trash", "trashed": True}


# ============ File Content Endpoints ============


@router.get("/files/{file_id}/download")
async def download_file(
    file_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Download a file."""
    user, can_read_all = auth_result
    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)

    abs_path = to_absolute_path(file.file_path)
    if not abs_path or not abs_path.exists():
        raise HTTPException(status_code=404, detail="File not found on disk")

    return FastAPIFileResponse(
        str(abs_path),
        filename=file.filename,
        media_type="application/octet-stream",
    )


@router.post("/files/{file_id}/slicer-token")
async def create_library_slicer_token(
    file_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Create a short-lived download token for opening files in slicer applications.

    Slicer protocol handlers (bambustudioopen://, orcaslicer://) cannot send
    auth headers, so they use this token in the URL path instead.
    """
    from backend.app.core.auth import create_slicer_download_token

    user, can_read_all = auth_result
    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)

    token = await create_slicer_download_token("library", file_id)
    return {"token": token}


@router.get("/files/{file_id}/dl/{token}/{filename}")
async def download_library_file_for_slicer(
    file_id: int,
    token: str,
    filename: str,
    db: AsyncSession = Depends(get_db, scope="function"),
):
    """Download a library file using a slicer download token.

    Token-authenticated (no auth headers needed). The token is short-lived and
    file-bound, created by POST /files/{file_id}/slicer-token, and redeemable
    for the rest of its TTL rather than exactly once -- the slicer is a separate
    process that may fetch the URL more than once (#3029).
    Filename is at the end of the URL so slicers can detect the file format.
    """
    from backend.app.core.auth import verify_slicer_download_token

    if not await verify_slicer_download_token(token, "library", file_id, single_use=False):
        raise HTTPException(status_code=403, detail="Invalid or expired download token")

    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    file = result.scalar_one_or_none()
    if not file:
        raise HTTPException(status_code=404, detail="File not found")

    abs_path = to_absolute_path(file.file_path)
    if not abs_path or not abs_path.exists():
        raise HTTPException(status_code=404, detail="File not found on disk")

    return FastAPIFileResponse(
        str(abs_path),
        filename=file.filename,
        media_type="application/octet-stream",
    )


@router.get("/files/{file_id}/thumbnail")
async def get_thumbnail(
    file_id: int,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_media_token_ownership(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Get a file's thumbnail.

    Accepts a media token in ``?token=`` because <img> cannot send headers.
    Ownership is enforced here rather than assumed from the credential: until
    #3025 this route took the anonymous camera-stream token, which carried no
    principal, so any holder could read any user's thumbnail by walking IDs.
    """
    user, can_read_all = auth_result
    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)

    abs_thumb_path = to_absolute_path(file.thumbnail_path)
    if not abs_thumb_path or not abs_thumb_path.exists():
        raise HTTPException(status_code=404, detail="Thumbnail not found")

    # Detect media type from extension
    thumb_ext = abs_thumb_path.suffix.lower()
    media_types = {
        ".png": "image/png",
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".gif": "image/gif",
        ".webp": "image/webp",
    }
    media_type = media_types.get(thumb_ext, "image/png")

    return FastAPIFileResponse(str(abs_thumb_path), media_type=media_type)


@router.post("/files/{file_id}/preview-thumbnail", response_model=ClientThumbnailResponse)
async def upload_preview_thumbnail(
    file_id: int,
    thumbnail: UploadFile = File(...),
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_UPDATE_ALL,
            Permission.LIBRARY_UPDATE_OWN,
        )
    ),
):
    """Store a client-rendered preview thumbnail for a file (#2976).

    STEP, PDF and spreadsheet previews are rendered in the browser; the FE
    posts its first render here so the grid gets a thumbnail without the
    server needing OpenCascade. A PDF normally has its PDFium thumbnail from
    upload already, so for PDFs this only fills the gap for a file PDFium
    could not read. Only file types in ``CLIENT_THUMBNAIL_TYPES`` are
    accepted, and only while the file has no thumbnail yet — a stored
    thumbnail is never replaced by this route.
    """
    user, can_modify_all = auth_result

    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    file = result.scalar_one_or_none()

    if not file:
        raise HTTPException(status_code=404, detail="File not found")

    # Ownership check (same shape as update_file)
    if not can_modify_all:
        if file.created_by_id != user.id:
            raise HTTPException(status_code=403, detail="You can only update your own files")

    if file.file_type not in CLIENT_THUMBNAIL_TYPES:
        raise HTTPException(status_code=400, detail="File type does not accept client-rendered thumbnails")

    if file.thumbnail_path:
        return ClientThumbnailResponse(updated=False)

    content = await thumbnail.read(MAX_CLIENT_THUMBNAIL_BYTES + 1)
    if len(content) > MAX_CLIENT_THUMBNAIL_BYTES:
        raise HTTPException(status_code=413, detail="Thumbnail too large")

    # Decode and re-encode through PIL: validates the bytes are a real PNG
    # and strips anything that isn't pixel data before it lands on disk.
    import io

    from PIL import Image, UnidentifiedImageError

    try:
        # Image.open() reads the header only. Both checks below happen before
        # load(), so a declared-but-never-delivered canvas is refused rather
        # than allocated. DecompressionBombError derives straight from
        # Exception, so it has to be named explicitly — open() itself raises
        # it once the declared size passes PIL's own limit.
        with Image.open(io.BytesIO(content)) as source:
            if source.format != "PNG":
                raise HTTPException(status_code=400, detail="Thumbnail must be a PNG image")
            if max(source.size) > MAX_CLIENT_THUMBNAIL_EDGE:
                raise HTTPException(status_code=400, detail="Thumbnail image dimensions too large")
            source.load()
            img = source.convert("RGBA") if source.mode not in ("RGB", "RGBA") else source.copy()
    except HTTPException:
        raise
    except (UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError) as e:
        raise HTTPException(status_code=400, detail="Invalid thumbnail image") from e

    if max(img.size) > STORED_CLIENT_THUMBNAIL_EDGE:
        img.thumbnail((STORED_CLIENT_THUMBNAIL_EDGE, STORED_CLIENT_THUMBNAIL_EDGE), Image.Resampling.LANCZOS)

    thumbnails_dir = get_library_thumbnails_dir()
    thumb_filename = f"{uuid.uuid4().hex}.png"
    thumb_path = thumbnails_dir / thumb_filename  # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + ".png"
    # Outside the decode guard on purpose: a full disk or an unwritable
    # thumbnail directory is ours, not "Invalid thumbnail image".
    try:
        img.save(thumb_path, "PNG", optimize=True)
    except OSError as e:
        logger.error("Failed to store preview thumbnail for file %s: %s", file_id, e)
        raise HTTPException(status_code=500, detail="Failed to store thumbnail") from e

    # Two previews of the same file can reach this point together; the loser
    # of the UPDATE takes its PNG back off disk instead of orphaning it.
    result = await db.execute(
        update(LibraryFile)
        .where(LibraryFile.id == file_id, LibraryFile.thumbnail_path.is_(None))
        .values(thumbnail_path=to_relative_path(thumb_path))
    )
    await db.commit()
    if result.rowcount == 0:
        thumb_path.unlink(missing_ok=True)
        return ClientThumbnailResponse(updated=False)

    return ClientThumbnailResponse(updated=True)


# ============ Photo Endpoints (#3077) ============


@router.post("/files/{file_id}/photos")
async def upload_file_photo(
    file_id: int,
    file: UploadFile = File(...),
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_UPDATE_ALL,
            Permission.LIBRARY_UPDATE_OWN,
        )
    ),
):
    """Attach a photo of the printed result to a library file.

    Photos are Bambuddy-side metadata, so external files take them too. Same
    shape as the archive photo upload: extension allowlist, uuid-named on
    disk, and the ``photos`` list re-assigned so SQLAlchemy sees the change.
    """
    user, can_modify_all = auth_result

    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    library_file = result.scalar_one_or_none()

    if not library_file:
        raise HTTPException(status_code=404, detail="File not found")

    # Ownership check (same shape as update_file)
    if not can_modify_all:
        if library_file.created_by_id != user.id:
            raise HTTPException(status_code=403, detail="You can only update your own files")

    if not file.filename or not file.filename.lower().endswith(PHOTO_EXTENSIONS):
        raise HTTPException(status_code=400, detail="File must be an image (.jpg, .jpeg, .png, .webp)")

    content = await file.read(MAX_PHOTO_BYTES + 1)
    if len(content) > MAX_PHOTO_BYTES:
        raise HTTPException(status_code=413, detail="Photo too large (max 10 MB)")

    photos_dir = library_photos_dir(library_file.id)
    photos_dir.mkdir(parents=True, exist_ok=True)

    ext = Path(file.filename).suffix.lower()
    photo_filename = f"{uuid.uuid4().hex[:8]}{ext}"
    photo_path = photos_dir / photo_filename  # SEC-PATH-OK: photo_filename = uuid.uuid4().hex[:8] + ext
    photo_path.write_bytes(content)

    photos = list(library_file.photos or [])
    photos.append(photo_filename)
    library_file.photos = photos

    await db.commit()
    await db.refresh(library_file)

    return {"status": "uploaded", "filename": photo_filename, "photos": library_file.photos}


@router.get("/files/{file_id}/photos/{filename}")
async def get_file_photo(
    file_id: int,
    filename: str,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_media_token_ownership(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Serve one photo. Media-token auth like the thumbnail route (#3025)."""
    user, can_read_all = auth_result
    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    library_file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)

    # Membership check first: names are uuid-generated on upload, so anything
    # not in the stored list is not a photo, whatever is on disk.
    if not library_file.photos or filename not in library_file.photos:
        raise HTTPException(status_code=404, detail="Photo not found")

    try:
        photo_path = safe_join_under(library_photos_dir(library_file.id), filename, http=False)
    except PathTraversalError:
        raise HTTPException(status_code=404, detail="Photo not found") from None
    if not photo_path.is_file():
        raise HTTPException(status_code=404, detail="Photo not found")

    media_type = PHOTO_MEDIA_TYPES.get(Path(filename).suffix.lower(), "image/jpeg")
    return FastAPIFileResponse(str(photo_path), media_type=media_type)


@router.delete("/files/{file_id}/photos/{filename}")
async def delete_file_photo(
    file_id: int,
    filename: str,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_UPDATE_ALL,
            Permission.LIBRARY_UPDATE_OWN,
        )
    ),
):
    """Remove a photo from a library file."""
    user, can_modify_all = auth_result

    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    library_file = result.scalar_one_or_none()

    if not library_file:
        raise HTTPException(status_code=404, detail="File not found")

    if not can_modify_all:
        if library_file.created_by_id != user.id:
            raise HTTPException(status_code=403, detail="You can only update your own files")

    if not library_file.photos or filename not in library_file.photos:
        raise HTTPException(status_code=404, detail="Photo not found")

    try:
        photo_path = safe_join_under(library_photos_dir(library_file.id), filename, http=False)
    except PathTraversalError:
        raise HTTPException(status_code=404, detail="Photo not found") from None
    if photo_path.is_file():
        try:
            photo_path.unlink()
        except OSError as e:
            logger.warning("Failed to delete photo from disk: %s", e)

    photos = [p for p in library_file.photos if p != filename]
    library_file.photos = photos if photos else None

    await db.commit()

    return {"status": "deleted", "photos": library_file.photos or []}


@router.get("/files/{file_id}/gcode")
async def get_gcode(
    file_id: int,
    plate: int | None = None,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Get gcode for a file (for preview).

    Mirrors the archive route: ``?plate=2`` returns ``Metadata/plate_2.gcode``,
    and omitting it returns the lowest-numbered plate. The viewer has been
    sending ``plate`` since it gained a multi-plate URL, but this route took no
    such parameter and FastAPI drops unknown query parameters silently — so
    every multi-plate library file opened on whichever plate the slicer wrote
    first into the zip, which is not plate 1.
    """
    user, can_read_all = auth_result
    result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
    file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)

    abs_path = to_absolute_path(file.file_path)
    if not abs_path or not abs_path.exists():
        raise HTTPException(status_code=404, detail="File not found on disk")

    # Legacy sliced rows from before #1709 stored a `.gcode.3mf` ZIP body
    # under file_type="gcode" — the on-disk filename is the truth in that
    # case, so detect by suffix before checking the type column.
    is_gcode_3mf = file.file_type in ("3mf", "gcode.3mf") or file.filename.lower().endswith(".gcode.3mf")

    if plate is not None and plate < 1:
        raise HTTPException(status_code=400, detail="Plate index must be >= 1")

    if is_gcode_3mf:
        try:
            with zipfile.ZipFile(str(abs_path), "r") as zf:
                gcode_files = [n for n in zf.namelist() if n.endswith(".gcode")]
                if not gcode_files:
                    raise HTTPException(status_code=404, detail="No gcode found in 3MF file")
                if plate is not None:
                    selected = select_plate_gcode_name(gcode_files, plate)
                    if selected is None:
                        raise HTTPException(status_code=404, detail=f"Plate {plate} not found in this file")
                else:
                    selected = default_plate_gcode_name(gcode_files)
                gcode_content = zf.read(selected)
                from fastapi.responses import Response

                return Response(content=gcode_content, media_type="text/plain")
        except zipfile.BadZipFile:
            raise HTTPException(status_code=400, detail="Invalid 3MF file")
    elif file.file_type == "gcode":
        return FastAPIFileResponse(str(abs_path), media_type="text/plain")
    else:
        raise HTTPException(status_code=400, detail="Unsupported file type")


# ============ Bulk Operations ============


@router.post("/files/move")
async def move_files(
    data: FileMoveRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_UPDATE_ALL,
            Permission.LIBRARY_UPDATE_OWN,
        )
    ),
):
    """Move multiple files to a folder.

    Cross-boundary moves (managed ↔ external, or external ↔ external)
    physically relocate the bytes — see ``_move_file_bytes``. Same-boundary
    moves stay DB-only because the file's on-disk location doesn't depend
    on which managed folder owns it.

    Files not owned by the user are skipped (unless user has ``*_all``
    permission). Each skip carries a structured reason so the UI can
    surface "5 of 10 files were skipped: 3 had filename collisions on
    the NAS, 2 are no longer on disk" rather than a blank "skipped: 5".
    """
    user, can_modify_all = auth_result

    # Verify folder exists if specified
    target_folder: LibraryFolder | None = None
    if data.folder_id is not None:
        # Only into a folder the user may write to (#3201); library:update_all
        # moves into any folder, as before.
        target_folder = await get_writable_folder(db, data.folder_id, None if can_modify_all else user)
        if target_folder.is_external and target_folder.external_readonly:
            raise HTTPException(status_code=403, detail="Cannot move files to a read-only external folder")

    target_is_external = target_folder is not None and target_folder.is_external

    moved = 0
    skipped = 0
    skipped_reasons: list[dict] = []

    for file_id in data.file_ids:
        result = await db.execute(
            LibraryFile.active().options(selectinload(LibraryFile.folder)).where(LibraryFile.id == file_id)
        )
        file = result.scalar_one_or_none()
        if not file:
            continue
        # Ownership check
        if not can_modify_all and file.created_by_id != user.id:
            skipped += 1
            skipped_reasons.append({"file_id": file_id, "code": "not_owner", "reason": "not the file owner"})
            continue

        # No bytes need to move when both ends are managed (same-boundary).
        if not file.is_external and not target_is_external:
            file.folder_id = data.folder_id
            moved += 1
            continue

        # Block moves out of a read-only external mount. The user only has
        # read access to the source, and a move is semantically a delete on
        # the source — which a read-only mount can't fulfil. Without this
        # guard we'd succeed at copying to the target, fail to unlink the
        # source, and the same file would now exist in two places (with
        # the DB pointing at only one).
        if file.is_external and file.folder is not None and file.folder.external_readonly:
            skipped += 1
            skipped_reasons.append(
                {"file_id": file_id, "code": "source_readonly", "reason": "source is on a read-only external folder"}
            )
            continue

        # Otherwise relocate the bytes, then update the DB row to match.
        try:
            new_file_path = _move_file_bytes(file, target_folder)
        except _MoveSkip as e:
            skipped += 1
            skipped_reasons.append({"file_id": file_id, "code": e.code, "reason": e.reason})
            continue

        file.is_external = target_is_external
        file.folder_id = data.folder_id
        file.file_path = new_file_path
        # External rows historically carry `file_hash=None` (scan skips
        # hashing). When pulling an external file into managed storage,
        # compute the hash so dedup detection works for future uploads
        # of the same content.
        if not target_is_external and file.file_hash is None:
            try:
                abs_path = to_absolute_path(new_file_path)
                if abs_path:
                    file.file_hash = calculate_file_hash(abs_path)
            except OSError:
                pass  # leave hash null; dedup just won't match this row
        moved += 1

    await db.commit()

    return {
        "status": "success",
        "moved": moved,
        "skipped": skipped,
        "skipped_reasons": skipped_reasons,
    }


@router.post("/bulk-delete", response_model=BulkDeleteResponse)
async def bulk_delete(
    data: BulkDeleteRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_DELETE_ALL,
            Permission.LIBRARY_DELETE_OWN,
        )
    ),
):
    """Delete multiple files and/or folders.

    Files not owned by the user are skipped (unless user has *_all permission).
    """
    from backend.app.services.library_trash import delete_dependent_variants, release_queue_references

    user, can_modify_all = auth_result
    deleted_files = 0
    deleted_folders = 0
    skipped_files = 0
    # External files bypass the trash and are removed for good, so the queue has
    # to come off them. Collected here and dealt with once, below the loop.
    hard_deleted: list[LibraryFile] = []

    # Delete files first. Managed files go to trash (sweeper hard-deletes bytes
    # later); external files bypass trash since their disk state is outside our
    # control and can't be restored from trash anyway.
    now = datetime.now(timezone.utc)
    for file_id in data.file_ids:
        result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
        file = result.scalar_one_or_none()
        if not file:
            continue
        if not can_modify_all and file.created_by_id != user.id:
            skipped_files += 1
            continue

        if file.is_external:
            abs_thumb_path = to_absolute_path(file.thumbnail_path)
            if abs_thumb_path and abs_thumb_path.exists():
                try:
                    abs_thumb_path.unlink()
                except OSError as e:
                    logger.warning("Failed to delete thumbnail from disk: %s", e)
            hard_deleted.append(file)
        else:
            file.deleted_at = now
        deleted_files += 1

    # After the loop and before any delete is issued (#2819). Order matters
    # twice over: a query run while a delete is pending autoflushes it, taking
    # the cascade with it, and releasing once for the whole set is a couple of
    # statements rather than a couple per file.
    if hard_deleted:
        hard_deleted_ids = [f.id for f in hard_deleted]
        await delete_dependent_variants(db, hard_deleted_ids)
        await release_queue_references(db, hard_deleted_ids)
        for file in hard_deleted:
            remove_library_photos_dir(file.id)
            await db.delete(file)

    # Delete folders (cascade will handle contents). Without *_all permission a
    # user deletes only what DELETE /folders/{id} would let them (#1781, #3201);
    # a folder they can't see is skipped like one that doesn't exist.
    index = await load_folder_index(db) if data.folder_ids and not can_modify_all else None
    visible = visible_folder_ids(index, user) if index is not None else None
    for folder_id in data.folder_ids:
        result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
        folder = result.scalar_one_or_none()
        if folder:
            if visible is not None and folder.id not in visible:
                continue
            if not can_modify_all and folder_delete_blocker(index, folder, user):
                continue
            # Count files that will be deleted
            file_count_result = await db.execute(
                select(func.count(LibraryFile.id)).where(
                    LibraryFile.folder_id == folder_id,
                    LibraryFile.deleted_at.is_(None),
                )
            )
            deleted_files += file_count_result.scalar() or 0
            tree_file_ids = await _folder_tree_file_ids(db, folder_id)
            await delete_dependent_variants(db, tree_file_ids)
            await release_queue_references(db, tree_file_ids)
            # The cascade hard-deletes every row in the subtree, so their
            # photos go with them — same as DELETE /folders/{id} (#3077).
            for doomed_id in tree_file_ids:
                remove_library_photos_dir(doomed_id)
            await db.delete(folder)
            deleted_folders += 1

    await db.commit()

    return BulkDeleteResponse(deleted_files=deleted_files, deleted_folders=deleted_folders)


# ============ Stats Endpoint ============


@router.get("/stats")
async def get_library_stats(
    db: AsyncSession = Depends(get_db, scope="function"),
    auth_result: tuple[User | ApiKeyActor | None, bool] = Depends(
        require_ownership_permission(
            Permission.LIBRARY_READ_ALL,
            Permission.LIBRARY_READ_OWN,
        )
    ),
):
    """Get library statistics."""
    user, can_read_all = auth_result
    # Stats exclude trashed files — users see counts/sizes for what's actually in the library.
    # Without LIBRARY_READ_ALL the stats reflect only the caller's own files —
    # match what the file list endpoint shows so the numbers stay consistent.
    file_filters = [LibraryFile.deleted_at.is_(None)]
    if user is not None and not can_read_all:
        file_filters.append(LibraryFile.created_by_id == user.id)

    # Total files
    total_files_result = await db.execute(select(func.count(LibraryFile.id)).where(*file_filters))
    total_files = total_files_result.scalar() or 0

    # Total folders: the ones the user can see (#3201)
    visible = None if sees_all_folders(user) else visible_folder_ids(await load_folder_index(db), user)
    if visible is None:
        total_folders_result = await db.execute(select(func.count(LibraryFolder.id)))
        total_folders = total_folders_result.scalar() or 0
    else:
        total_folders = len(visible)

    # Total size
    total_size_result = await db.execute(select(func.sum(LibraryFile.file_size)).where(*file_filters))
    total_size = total_size_result.scalar() or 0

    # Files by type
    type_result = await db.execute(
        select(LibraryFile.file_type, func.count(LibraryFile.id)).where(*file_filters).group_by(LibraryFile.file_type)
    )
    files_by_type = dict(type_result.all())

    # Total prints
    total_prints_result = await db.execute(select(func.sum(LibraryFile.print_count)).where(*file_filters))
    total_prints = total_prints_result.scalar() or 0

    # Disk space info
    library_dir = get_library_dir()
    try:
        disk_stat = shutil.disk_usage(library_dir)
        disk_free_bytes = disk_stat.free
        disk_total_bytes = disk_stat.total
        disk_used_bytes = disk_stat.used
    except OSError:
        disk_free_bytes = 0
        disk_total_bytes = 0
        disk_used_bytes = 0

    return {
        "total_files": total_files,
        "total_folders": total_folders,
        "total_size_bytes": total_size,
        "files_by_type": files_by_type,
        "total_prints": total_prints,
        "disk_free_bytes": disk_free_bytes,
        "disk_total_bytes": disk_total_bytes,
        "disk_used_bytes": disk_used_bytes,
    }