feat(slicer-plugin): upload API for slicer plugins

Adds GET /slicer-plugin/info and POST /slicer-plugin/uploads, which take
a sliced file plus a JSON manifest in one multipart request: what the user
asked for (store, or queue a plate x copies on a printer, optionally held
for a manual start) and where the file came from (slicer and plugin
versions, presets, a model fingerprint kept for later matching).

The manifest does not carry the slice's numbers; plates, time, grams and
sliced_for_model are read from the file by the library's own parser. The
work is delegated rather than re-implemented: the body of POST
/library/files becomes a shared ingest_library_upload, and queueing calls
POST /queue/ itself, so every validation, compatibility, filament and
budget check applies unchanged.

Sends are idempotent on a plugin-chosen upload_id, claimed through a
unique constraint, released when the attempt fails, taken over when
abandoned, and replayed only to the caller who first used it; a retried
queue send is otherwise a second print. A missing plate and a key that may
not queue are refused before anything is stored. A refusal from the queue
keeps the file, reports queue_error and rolls back what that route wrote
first, since with billing on it creates the batch before asking for a cost
centre. cost_center_id is accepted for the same reason.

New table slicer_plugin_uploads, created on startup.

backend/app/api/routes/library.py

@@ -2235,162 +2235,189 @@ async def list_files(
     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),
-    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
-):
-    """Upload a file to the library."""
+async def ingest_library_upload(
+    db: AsyncSession,
+    *,
+    filename: str,
+    content: bytes,
+    folder_id: int | None,
+    created_by_id: int | None,
+    generate_stl_thumbnails: bool = True,
+) -> tuple[LibraryFile, int | None]:
+    """Validate, store and index one uploaded file as a library row, and commit.
+
+    Everything ``POST /library/files`` does once the request is read. Shared
+    with the slicer plugin's upload (``slicer_plugin.py``) so a file sent from
+    the slicer is checked, hashed, parsed and thumbnailed exactly as one
+    dropped into the library is -- a second copy of these rules would be a
+    second thing to keep in step. Raises HTTPException for the caller's
+    mistakes. Returns the new row and the id of an active file with the same
+    bytes, if there is one.
+    """
+    # Reject FAT32/exFAT-incompatible filenames up front (#1540).
     try:
-        if not file.filename:
-            raise HTTPException(status_code=400, detail="Filename is required")
+        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()
 
-        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
-        target_folder = 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()
-            if not target_folder:
-                raise HTTPException(status_code=404, detail="Folder not found")
+    # Verify folder exists if specified
+    target_folder = 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()
+        if not target_folder:
+            raise HTTPException(status_code=404, detail="Folder not found")
 
-        # 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)
+    # 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)
+    # The caller has read the upload so the validation can sniff magic bytes;
+    # the file is written to disk only after the checks. #1401.
+    validate_print_file_upload(filename, content)
 
-        # Save file
-        with open(file_path, "wb") as f:
-            f.write(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)
+    # 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)
+    # 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()
+    # 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()
+    # 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
+    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)
 
-                metadata = clean_metadata(raw_metadata)
-            except Exception as e:
-                logger.warning("Failed to parse 3MF: %s", e)
+            # 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
+    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 == ".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:
-                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 == ".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 = generate_stl_thumbnail(file_path, thumbnails_dir)
-                except OSError:
-                    pass
+                if file_path.stat().st_size >= MIN_USABLE_STL_BYTES:
+                    thumbnail_path = generate_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(
+    # 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=created_by_id,
+    )
+    db.add(library_file)
+    await db.commit()
+    await db.refresh(library_file)
+
+    return library_file, duplicate_of
+
+
+@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),
+    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
+):
+    """Upload a file to the library."""
+    try:
+        if not file.filename:
+            raise HTTPException(status_code=400, detail="Filename is required")
+
+        library_file, duplicate_of = await ingest_library_upload(
+            db,
+            filename=file.filename,
+            content=await file.read(),
             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=current_user.id if current_user else None,
+            generate_stl_thumbnails=generate_stl_thumbnails,
         )
-        db.add(library_file)
-        await db.commit()
-        await db.refresh(library_file)
 
         return FileUploadResponse(
             id=library_file.id,

backend/app/api/routes/slicer_plugin.py

@@ -0,0 +1,332 @@
+"""API for the Bambuddy slicer plugin (OrcaSlicer).
+
+The plugin runs inside the slicer on the user's machine and sends a sliced
+file here together with a manifest: what the user wants done with it (library
+only, or queue plate N x copies on a printer) and where it came from (slicer,
+presets, a fingerprint of the model). The file's own contents -- per-plate
+grams, time, the model it was sliced for -- are read from the file by the same
+parser every library upload goes through, not taken from the manifest, so the
+two can never disagree.
+
+The work is delegated: storing the file is ``ingest_library_upload`` and
+queueing it is ``POST /queue/`` itself, so a file sent from the slicer passes
+exactly the checks one added through the UI does.
+"""
+
+import asyncio
+import io
+import json
+import logging
+import zipfile
+from datetime import datetime, timedelta, timezone
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, File, Form, Header, HTTPException, UploadFile
+from fastapi.security import HTTPAuthorizationCredentials
+from pydantic import ValidationError
+from sqlalchemy import delete, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.api.routes.library import ingest_library_upload
+from backend.app.api.routes.print_queue import add_to_queue
+from backend.app.core.auth import _validate_api_key, require_permission_if_auth_enabled, security
+from backend.app.core.config import APP_VERSION
+from backend.app.core.database import get_db
+from backend.app.core.permissions import Permission
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.models.slicer_plugin_upload import SlicerPluginUpload
+from backend.app.models.user import User
+from backend.app.schemas.print_queue import PrintQueueItemCreate
+from backend.app.schemas.slicer_plugin import (
+    SlicerPluginInfoResponse,
+    SlicerPluginLibraryFile,
+    SlicerPluginManifest,
+    SlicerPluginQueueResult,
+    SlicerPluginUploadResponse,
+)
+from backend.app.utils.printer_models import is_gcode_compatible
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/slicer-plugin", tags=["slicer-plugin"])
+
+# Manifest versions this server understands. A plugin reads them from /info and
+# sends the newest one both sides know.
+MANIFEST_VERSIONS = (1,)
+
+# A claim still "processing" after this long belongs to a request that died
+# without releasing it (a killed worker, a cancelled task). Past it, a retry
+# with the same upload_id may take the claim over instead of being refused
+# forever. Far longer than any upload plus queue insert takes.
+_STALE_CLAIM = timedelta(minutes=10)
+
+_MAX_MANIFEST_CHARS = 64_000
+
+
+def _utcnow() -> datetime:
+    return datetime.now(timezone.utc).replace(tzinfo=None)
+
+
+async def _may_queue(credentials: HTTPAuthorizationCredentials | None, x_api_key: str | None) -> bool:
+    """Whether the caller may create queue items, decided by the queue's own gate.
+
+    Asks the dependency ``POST /queue/`` is guarded by rather than restating
+    its rules, so a key or account that could not queue from the UI cannot
+    queue from the slicer either.
+    """
+    try:
+        await require_permission_if_auth_enabled(Permission.QUEUE_CREATE)(credentials=credentials, x_api_key=x_api_key)
+    except HTTPException as e:
+        if e.status_code == 403:
+            return False
+        raise
+    return True
+
+
+async def _caller_id(
+    db: AsyncSession,
+    user: User | None,
+    credentials: HTTPAuthorizationCredentials | None,
+    x_api_key: str | None,
+) -> str:
+    """Name the caller, so an upload_id replays only to whoever used it first."""
+    if user is not None:
+        return f"user:{user.id}"
+    key_value = x_api_key
+    if key_value is None and credentials is not None and credentials.credentials.startswith("bb_"):
+        key_value = credentials.credentials
+    if key_value:
+        api_key = await _validate_api_key(db, key_value)
+        if api_key is not None:
+            return f"apikey:{api_key.id}"
+    return "anonymous"
+
+
+def _check_plate(content: bytes, plate_id: int) -> str | None:
+    """Return why ``plate_id`` cannot be printed from this file, or None if it can."""
+    try:
+        with zipfile.ZipFile(io.BytesIO(content)) as zf:
+            names = set(zf.namelist())
+    except zipfile.BadZipFile:
+        return "plate_id needs a sliced 3MF; this file is not one"
+    if f"Metadata/plate_{plate_id}.gcode" not in names:
+        return f"This file has no sliced plate {plate_id}"
+    return None
+
+
+def _replay(existing: SlicerPluginUpload, caller: str) -> SlicerPluginUploadResponse:
+    """Answer a repeated upload_id with what the first request returned."""
+    # Anyone else presenting the id gets nothing about it: a 409 that is the
+    # same whether or not the stored upload is still in progress.
+    if existing.caller != caller:
+        raise HTTPException(409, "upload_id has already been used")
+    if existing.status != "done" or not existing.response:
+        raise HTTPException(409, "An upload with this upload_id is still being processed")
+    data = json.loads(existing.response)
+    data["replayed"] = True
+    return SlicerPluginUploadResponse.model_validate(data)
+
+
+@router.get("/info", response_model=SlicerPluginInfoResponse)
+async def plugin_info(
+    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
+    credentials: HTTPAuthorizationCredentials | None = Depends(security),
+    x_api_key: Annotated[str | None, Header(alias="X-API-Key")] = None,
+):
+    """Handshake for the plugin's connect dialog.
+
+    Needs the same permission an upload does, so "connected" means the plugin
+    can actually send. ``can_queue`` tells it whether to offer queueing at all.
+    """
+    return SlicerPluginInfoResponse(
+        bambuddy_version=APP_VERSION,
+        manifest_versions=list(MANIFEST_VERSIONS),
+        can_upload=True,
+        can_queue=await _may_queue(credentials, x_api_key),
+    )
+
+
+@router.post("/uploads", response_model=SlicerPluginUploadResponse)
+async def upload_from_slicer(
+    file: UploadFile = File(...),
+    manifest: str = Form(..., max_length=_MAX_MANIFEST_CHARS),
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
+    credentials: HTTPAuthorizationCredentials | None = Depends(security),
+    x_api_key: Annotated[str | None, Header(alias="X-API-Key")] = None,
+):
+    """Store a sliced file from the slicer plugin, and queue it if asked.
+
+    Idempotent on ``manifest.upload_id``: a retry of a send whose response was
+    lost gets the first response back (``replayed: true``) instead of a second
+    upload -- and, with action "queue", instead of a second print.
+    """
+    try:
+        m = SlicerPluginManifest.model_validate_json(manifest)
+    except ValidationError as e:
+        raise HTTPException(422, detail=e.errors(include_url=False, include_context=False, include_input=False))
+    if m.schema_version not in MANIFEST_VERSIONS:
+        raise HTTPException(
+            400,
+            f"Manifest version {m.schema_version} is not supported; this server reads {list(MANIFEST_VERSIONS)}",
+        )
+    if not file.filename:
+        raise HTTPException(400, "Filename is required")
+
+    intent = m.intent
+    caller = await _caller_id(db, current_user, credentials, x_api_key)
+
+    # A repeat is answered before anything is checked or read, so a retry of an
+    # upload that succeeded returns its answer even if the caller has since
+    # lost a permission it had then.
+    existing = (
+        await db.execute(select(SlicerPluginUpload).where(SlicerPluginUpload.upload_id == m.upload_id))
+    ).scalar_one_or_none()
+    if existing is not None:
+        abandoned = (
+            existing.status != "done" and existing.caller == caller and existing.created_at < _utcnow() - _STALE_CLAIM
+        )
+        if not abandoned:
+            return _replay(existing, caller)
+        await db.delete(existing)
+        await db.commit()
+
+    # Refused before anything is written: a plugin told "uploaded, but you may
+    # not queue" would leave a file in the library the user never asked to keep.
+    if intent.action == "queue" and not await _may_queue(credentials, x_api_key):
+        raise HTTPException(403, "This account or API key may upload files but not queue them")
+
+    content = await file.read()
+    if intent.action == "queue" and intent.plate_id is not None:
+        plate_problem = await asyncio.to_thread(_check_plate, content, intent.plate_id)
+        if plate_problem:
+            raise HTTPException(400, plate_problem)
+
+    # Claim the id. The unique constraint decides between two requests that
+    # both got past the lookup above -- the plugin retrying a send that timed
+    # out while the first attempt was still working.
+    source = m.source
+    claim = SlicerPluginUpload(
+        upload_id=m.upload_id,
+        caller=caller,
+        status="processing",
+        slicer=source.slicer,
+        slicer_version=source.slicer_version,
+        plugin_version=source.plugin_version,
+        project_name=source.project_name,
+        model_fingerprint=source.model_fingerprint,
+        presets=source.presets.model_dump_json() if source.presets else None,
+        created_at=_utcnow(),
+    )
+    db.add(claim)
+    try:
+        await db.commit()
+    except IntegrityError:
+        await db.rollback()
+        winner = (
+            await db.execute(select(SlicerPluginUpload).where(SlicerPluginUpload.upload_id == m.upload_id))
+        ).scalar_one()
+        return _replay(winner, caller)
+    claim_id = claim.id
+
+    try:
+        response = await _store_and_queue(db, m, file.filename, content, current_user)
+    except Exception:
+        # Release the claim, so the plugin's retry runs again rather than
+        # being told the failed attempt is still in progress.
+        await db.rollback()
+        await db.execute(delete(SlicerPluginUpload).where(SlicerPluginUpload.id == claim_id))
+        await db.commit()
+        raise
+
+    claim = await db.get(SlicerPluginUpload, claim_id)
+    claim.status = "done"
+    claim.library_file_id = response.library_file.id
+    claim.response = response.model_dump_json()
+    await db.commit()
+    return response
+
+
+async def _store_and_queue(
+    db: AsyncSession,
+    m: SlicerPluginManifest,
+    filename: str,
+    content: bytes,
+    current_user: User | None,
+) -> SlicerPluginUploadResponse:
+    intent = m.intent
+    library_file, duplicate_of = await ingest_library_upload(
+        db,
+        filename=filename,
+        content=content,
+        folder_id=intent.folder_id,
+        created_by_id=current_user.id if current_user else None,
+    )
+    stored = SlicerPluginLibraryFile(
+        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,
+    )
+    sliced_for = (library_file.file_metadata or {}).get("sliced_for_model")
+
+    warnings: list[str] = []
+    if duplicate_of is not None:
+        warnings.append(f"The library already holds this exact file (#{duplicate_of}).")
+
+    if intent.action != "queue":
+        return SlicerPluginUploadResponse(upload_id=m.upload_id, library_file=stored, warnings=warnings)
+
+    # POST /queue/ refuses a target_model the file was not sliced for, but a
+    # named printer is the user's own pick and is only flagged -- the same
+    # latitude the print dialog gives.
+    if intent.printer_id is not None:
+        printer = await db.get(Printer, intent.printer_id)
+        if printer is not None and not is_gcode_compatible(sliced_for, printer.model):
+            warnings.append(f"Sliced for {sliced_for}, but {printer.name} is a {printer.model}.")
+
+    try:
+        item = await add_to_queue(
+            PrintQueueItemCreate(
+                library_file_id=library_file.id,
+                printer_id=intent.printer_id,
+                target_model=intent.target_model,
+                plate_id=intent.plate_id,
+                quantity=intent.copies,
+                manual_start=intent.manual_start,
+                cost_center_id=intent.cost_center_id,
+            ),
+            db=db,
+            current_user=current_user,
+        )
+    except HTTPException as e:
+        # The queue route may have flushed a batch row before refusing (the
+        # budget check runs after it); none of that may outlive the refusal.
+        await db.rollback()
+        queue_error = e.detail if isinstance(e.detail, str) else json.dumps(e.detail)
+        return SlicerPluginUploadResponse(
+            upload_id=m.upload_id, library_file=stored, queue_error=queue_error, warnings=warnings
+        )
+
+    if item.batch_id is None:
+        item_ids = [item.id]
+    else:
+        item_ids = list(
+            (
+                await db.execute(
+                    select(PrintQueueItem.id)
+                    .where(PrintQueueItem.batch_id == item.batch_id)
+                    .order_by(PrintQueueItem.position)
+                )
+            ).scalars()
+        )
+    return SlicerPluginUploadResponse(
+        upload_id=m.upload_id,
+        library_file=stored,
+        queue=SlicerPluginQueueResult(queue_item_ids=item_ids, batch_id=item.batch_id),
+        warnings=warnings,
+    )

backend/app/core/database.py

@@ -324,6 +324,7 @@ async def init_db():
         settings,
         shopping_list,
         slicer_pipeline,
+        slicer_plugin_upload,
         slot_preset,
         smart_plug,
         smart_plug_energy_snapshot,

backend/app/main.py

@@ -64,6 +64,7 @@ from backend.app.api.routes import (
     settings as settings_routes,
     slice_jobs,
     slicer_pipelines,
+    slicer_plugin,
     slicer_presets,
     smart_plugs,
     sponsor_prompt,
@@ -9904,6 +9905,7 @@ app.include_router(library_trash.router, prefix=app_settings.api_prefix)
 app.include_router(library_variants.router, prefix=app_settings.api_prefix)
 app.include_router(slice_jobs.router, prefix=app_settings.api_prefix)
 app.include_router(slicer_pipelines.router, prefix=app_settings.api_prefix)
+app.include_router(slicer_plugin.router, prefix=app_settings.api_prefix)
 app.include_router(pipeline_runs.pipeline_run_create_router, prefix=app_settings.api_prefix)
 app.include_router(pipeline_runs.pipeline_run_router, prefix=app_settings.api_prefix)
 app.include_router(slicer_presets.router, prefix=app_settings.api_prefix)

backend/app/models/__init__.py

@@ -28,6 +28,7 @@ from backend.app.models.project import Project
 from backend.app.models.scheduled_drying import ScheduledDrying
 from backend.app.models.settings import Settings
 from backend.app.models.slicer_pipeline import SlicerPipeline
+from backend.app.models.slicer_plugin_upload import SlicerPluginUpload
 from backend.app.models.smart_plug import SmartPlug
 from backend.app.models.smart_plug_energy_snapshot import SmartPlugEnergySnapshot
 from backend.app.models.sponsor_toast_state import SponsorToastState
@@ -82,6 +83,7 @@ __all__ = [
     "PipelineJob",
     "PipelineRun",
     "SlicerPipeline",
+    "SlicerPluginUpload",
     "Spool",
     "SpoolFilamentPreset",
     "SpoolKProfile",

backend/app/models/slicer_plugin_upload.py

@@ -0,0 +1,53 @@
+"""Uploads sent by the slicer plugin, and where each one came from."""
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.app.core.database import Base
+
+
+class SlicerPluginUpload(Base):
+    """One upload from the slicer plugin, keyed by the id the plugin chose for it.
+
+    Two jobs. It makes the upload idempotent: the plugin retries a send whose
+    response it never saw, and a retry that queued the file a second time would
+    print it twice. And it keeps what the file cannot say about itself -- which
+    slicer and presets produced it, and a fingerprint of the model -- which is
+    what lets a later slice of the same model find this one.
+    """
+
+    __tablename__ = "slicer_plugin_uploads"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+
+    # Chosen by the plugin, one per send. Unique, so two requests carrying the
+    # same id cannot both get past the claim in slicer_plugin.py.
+    upload_id: Mapped[str] = mapped_column(String(64), unique=True, index=True)
+
+    # Who sent it: "user:<id>", "apikey:<id>", or "anonymous" with auth off. A
+    # replay from anyone else is refused rather than answered with this
+    # caller's result.
+    caller: Mapped[str] = mapped_column(String(40))
+
+    # "processing" while the request that claimed the id is still working,
+    # "done" once `response` holds what it returned.
+    status: Mapped[str] = mapped_column(String(20), default="processing")
+
+    library_file_id: Mapped[int | None] = mapped_column(
+        ForeignKey("library_files.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+
+    # Provenance, as the plugin reported it.
+    slicer: Mapped[str | None] = mapped_column(String(50), nullable=True)
+    slicer_version: Mapped[str | None] = mapped_column(String(50), nullable=True)
+    plugin_version: Mapped[str | None] = mapped_column(String(50), nullable=True)
+    project_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    model_fingerprint: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    presets: Mapped[str | None] = mapped_column(Text, nullable=True)  # JSON
+
+    # The JSON body the first request returned, sent back verbatim on a replay.
+    response: Mapped[str | None] = mapped_column(Text, nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

backend/app/schemas/slicer_plugin.py

@@ -0,0 +1,95 @@
+"""Schemas for the slicer plugin's upload API."""
+
+from typing import Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+
+class SlicerPluginPresets(BaseModel):
+    """The preset names the slicer had selected when it sliced."""
+
+    printer: str | None = Field(default=None, max_length=255)
+    process: str | None = Field(default=None, max_length=255)
+    filaments: list[str] = Field(default_factory=list, max_length=64)
+
+
+class SlicerPluginSource(BaseModel):
+    """Where the file came from -- what the file itself cannot tell us."""
+
+    slicer: str | None = Field(default=None, max_length=50)
+    slicer_version: str | None = Field(default=None, max_length=50)
+    plugin_version: str | None = Field(default=None, max_length=50)
+    project_name: str | None = Field(default=None, max_length=255)
+    presets: SlicerPluginPresets | None = None
+    # A hash of the model's geometry, so a later slice of the same model can
+    # find this one however its file is named.
+    model_fingerprint: str | None = Field(default=None, max_length=128, pattern=r"^[A-Za-z0-9:_-]+$")
+
+
+class SlicerPluginIntent(BaseModel):
+    """What the user asked for when they sent the file."""
+
+    action: Literal["library", "queue"] = "library"
+    folder_id: int | None = None
+    # Queue only. 1-based, as in the 3MF (Metadata/plate_<n>.gcode).
+    plate_id: int | None = Field(default=None, ge=1)
+    copies: int = Field(default=1, ge=1, le=100)
+    printer_id: int | None = None
+    target_model: str | None = Field(default=None, max_length=50)
+    manual_start: bool = False
+    # Required by POST /queue/ when billing is enabled; ignored otherwise.
+    cost_center_id: int | None = None
+
+    @model_validator(mode="after")
+    def _one_destination(self) -> "SlicerPluginIntent":
+        if self.printer_id is not None and self.target_model:
+            raise ValueError("Cannot specify both printer_id and target_model")
+        return self
+
+
+class SlicerPluginManifest(BaseModel):
+    """The JSON part sent alongside the file.
+
+    Unknown keys are ignored so an older Bambuddy accepts a newer plugin's
+    additions; a change it must understand bumps ``schema_version`` instead.
+    """
+
+    schema_version: int
+    # Chosen by the plugin, one per send; a retry reuses it.
+    upload_id: str = Field(min_length=8, max_length=64, pattern=r"^[A-Za-z0-9-]+$")
+    source: SlicerPluginSource = Field(default_factory=SlicerPluginSource)
+    intent: SlicerPluginIntent = Field(default_factory=SlicerPluginIntent)
+
+
+class SlicerPluginLibraryFile(BaseModel):
+    id: int
+    filename: str
+    file_type: str
+    file_size: int
+    thumbnail_path: str | None
+    duplicate_of: int | None = None  # an active library file with the same bytes
+
+
+class SlicerPluginQueueResult(BaseModel):
+    queue_item_ids: list[int]
+    batch_id: int | None = None  # set when copies > 1
+
+
+class SlicerPluginUploadResponse(BaseModel):
+    upload_id: str
+    library_file: SlicerPluginLibraryFile
+    # With action "queue": what was queued, or why nothing was. The file is in
+    # the library either way, so the user can still queue it from there.
+    queue: SlicerPluginQueueResult | None = None
+    queue_error: str | None = None
+    warnings: list[str] = []
+    # True when this is the stored answer to an earlier request with the same
+    # upload_id, not a new upload.
+    replayed: bool = False
+
+
+class SlicerPluginInfoResponse(BaseModel):
+    bambuddy_version: str
+    manifest_versions: list[int]
+    can_upload: bool
+    can_queue: bool

backend/tests/integration/test_slicer_plugin_api.py

@@ -0,0 +1,380 @@
+"""The slicer plugin's upload API (``api/routes/slicer_plugin.py``).
+
+The route is a thin layer over two things with their own tests -- the library
+ingest and ``POST /queue/`` -- so these pin what the layer itself adds: the
+manifest, the idempotency claim, the up-front refusals that must leave nothing
+behind, and the permission split between uploading and queueing.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import uuid
+import zipfile
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app.core.auth import generate_api_key
+from backend.app.models.api_key import APIKey
+from backend.app.models.library import LibraryFile
+from backend.app.models.print_batch import PrintBatch
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.settings import Settings
+from backend.app.models.slicer_plugin_upload import SlicerPluginUpload
+from backend.app.models.user import User
+
+UPLOADS = "/api/v1/slicer-plugin/uploads"
+INFO = "/api/v1/slicer-plugin/info"
+
+
+def _sliced_3mf(plates=(1,), model_id: str = "C11") -> bytes:
+    """A minimal sliced 3MF: one G-code member per plate, and a slice_info that
+    names the printer it was sliced for (C11 = X1C)."""
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
+        for n in plates:
+            zf.writestr(f"Metadata/plate_{n}.gcode", "; G-code\nG28\n")
+        zf.writestr(
+            "Metadata/slice_info.config",
+            f'<config><plate><metadata key="index" value="{plates[0]}"/>'
+            f'<metadata key="printer_model_id" value="{model_id}"/></plate></config>',
+        )
+    return buf.getvalue()
+
+
+def _manifest(upload_id: str | None = None, **intent) -> dict:
+    return {
+        "schema_version": 1,
+        "upload_id": upload_id or str(uuid.uuid4()),
+        "source": {
+            "slicer": "OrcaSlicer",
+            "slicer_version": "2.4.0-nightly",
+            "plugin_version": "0.1.0",
+            "project_name": "Bracket v3",
+            "presets": {
+                "printer": "Bambu Lab X1 Carbon 0.4 nozzle",
+                "process": "0.20mm Standard",
+                "filaments": ["PLA"],
+            },
+            "model_fingerprint": "sha256:" + "a" * 64,
+        },
+        "intent": intent,
+    }
+
+
+async def _send(
+    client: AsyncClient, manifest: dict, content: bytes | None = None, headers=None, name="Bracket.gcode.3mf"
+):
+    return await client.post(
+        UPLOADS,
+        files={"file": (name, content if content is not None else _sliced_3mf(), "application/zip")},
+        data={"manifest": json.dumps(manifest)},
+        headers=headers or {},
+    )
+
+
+async def _count(test_engine, model, *where) -> int:
+    """Read through a fresh session -- the fixture's own can look stale after a
+    route call commits through its own ``get_db`` session."""
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    async with maker() as fresh:
+        return (await fresh.execute(select(func.count()).select_from(model).where(*where))).scalar_one()
+
+
+async def _claims(test_engine) -> list[SlicerPluginUpload]:
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    async with maker() as fresh:
+        return list((await fresh.execute(select(SlicerPluginUpload))).scalars())
+
+
+async def _admin_key(client: AsyncClient, db_session, **scopes) -> str:
+    """Turn auth on and mint a key owned by the admin, so the key's scope flags
+    alone decide what it may do."""
+    await client.post(
+        "/api/v1/auth/setup",
+        json={"auth_enabled": True, "admin_username": "plugadmin", "admin_password": "PlugPass1!"},
+    )
+    admin = (await db_session.execute(select(User).where(User.username == "plugadmin"))).scalar_one()
+    flags = {"can_manage_library": True, "can_queue": True, "can_read_status": True}
+    flags.update(scopes)
+    full_key, key_hash, key_prefix = generate_api_key()
+    db_session.add(
+        APIKey(
+            name=f"plugin-{key_prefix}",
+            key_hash=key_hash,
+            key_prefix=key_prefix,
+            enabled=True,
+            user_id=admin.id,
+            **flags,
+        )
+    )
+    await db_session.commit()
+    return full_key
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+class TestLibraryUpload:
+    async def test_stores_the_file_and_its_provenance(self, async_client, test_engine):
+        manifest = _manifest()
+        response = await _send(async_client, manifest)
+
+        assert response.status_code == 200, response.text
+        body = response.json()
+        assert body["replayed"] is False
+        assert body["queue"] is None and body["queue_error"] is None
+        assert body["library_file"]["filename"] == "Bracket.gcode.3mf"
+
+        [claim] = await _claims(test_engine)
+        assert claim.upload_id == manifest["upload_id"]
+        assert claim.status == "done"
+        assert claim.caller == "anonymous"
+        assert claim.library_file_id == body["library_file"]["id"]
+        assert claim.model_fingerprint == manifest["source"]["model_fingerprint"]
+        assert json.loads(claim.presets)["process"] == "0.20mm Standard"
+        assert claim.project_name == "Bracket v3"
+
+    async def test_a_retry_returns_the_first_answer_instead_of_a_second_file(self, async_client, test_engine):
+        manifest = _manifest()
+        first = (await _send(async_client, manifest)).json()
+        second = await _send(async_client, manifest)
+
+        assert second.status_code == 200
+        assert second.json()["replayed"] is True
+        assert second.json()["library_file"] == first["library_file"]
+        assert await _count(test_engine, LibraryFile) == 1
+
+    async def test_reports_an_identical_file_already_in_the_library(self, async_client):
+        first = (await _send(async_client, _manifest())).json()
+        second = (await _send(async_client, _manifest())).json()
+
+        assert second["library_file"]["duplicate_of"] == first["library_file"]["id"]
+        assert any(f"#{first['library_file']['id']}" in w for w in second["warnings"])
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+class TestQueueUpload:
+    async def test_queues_the_chosen_plate_as_a_batch_of_copies(self, async_client, printer_factory, test_engine):
+        printer = await printer_factory(model="X1C")
+        response = await _send(
+            async_client,
+            _manifest(action="queue", printer_id=printer.id, plate_id=2, copies=3, manual_start=True),
+            _sliced_3mf(plates=(1, 2)),
+        )
+
+        assert response.status_code == 200, response.text
+        queue = response.json()["queue"]
+        assert len(queue["queue_item_ids"]) == 3
+        assert queue["batch_id"] is not None
+
+        maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+        async with maker() as fresh:
+            items = list(
+                (
+                    await fresh.execute(select(PrintQueueItem).where(PrintQueueItem.id.in_(queue["queue_item_ids"])))
+                ).scalars()
+            )
+        assert {(i.printer_id, i.plate_id, i.manual_start, i.batch_id) for i in items} == {
+            (printer.id, 2, True, queue["batch_id"])
+        }
+
+    async def test_a_retry_does_not_queue_the_print_again(self, async_client, printer_factory, test_engine):
+        """The reason the endpoint is idempotent at all: a send whose response
+        was lost is retried, and a second queue insert is a second print."""
+        printer = await printer_factory(model="X1C")
+        manifest = _manifest(action="queue", printer_id=printer.id, copies=2)
+        first = (await _send(async_client, manifest)).json()
+        second = (await _send(async_client, manifest)).json()
+
+        assert second["replayed"] is True
+        assert second["queue"] == first["queue"]
+        assert await _count(test_engine, PrintQueueItem) == 2
+
+    async def test_a_plate_the_file_does_not_have_is_refused_before_anything_is_stored(
+        self, async_client, printer_factory, test_engine
+    ):
+        printer = await printer_factory(model="X1C")
+        response = await _send(async_client, _manifest(action="queue", printer_id=printer.id, plate_id=3))
+
+        assert response.status_code == 400
+        assert "plate 3" in response.json()["detail"]
+        assert await _count(test_engine, LibraryFile) == 0
+        assert await _claims(test_engine) == []
+
+    async def test_a_refused_queue_keeps_the_file_and_says_why(self, async_client, test_engine):
+        """No active printer of the target model: POST /queue/ refuses. The file
+        stays in the library -- the user can still queue it from there -- and
+        nothing of the half-made queue insert survives."""
+        response = await _send(async_client, _manifest(action="queue", target_model="H2D", copies=2))
+
+        assert response.status_code == 200, response.text
+        body = response.json()
+        assert body["queue"] is None
+        assert "No active printers" in body["queue_error"]
+        assert await _count(test_engine, LibraryFile) == 1
+        assert await _count(test_engine, PrintQueueItem) == 0
+        assert await _count(test_engine, PrintBatch) == 0
+
+    async def test_a_refusal_after_the_batch_was_made_leaves_no_batch_behind(
+        self, async_client, db_session, printer_factory, test_engine
+    ):
+        """With billing on and no cost centre, POST /queue/ refuses only after it
+        has already written the batch for the copies. Without the rollback the
+        request's own commit would keep that batch: a batch of nothing."""
+        db_session.add(Settings(key="billing_enabled", value="true"))
+        await db_session.commit()
+        printer = await printer_factory(model="X1C")
+
+        body = (await _send(async_client, _manifest(action="queue", printer_id=printer.id, copies=2))).json()
+
+        assert "Cost center is required" in body["queue_error"]
+        assert await _count(test_engine, PrintBatch) == 0
+        assert await _count(test_engine, PrintQueueItem) == 0
+
+    async def test_the_cost_center_reaches_the_queue(self, async_client, db_session, printer_factory):
+        db_session.add(Settings(key="billing_enabled", value="true"))
+        await db_session.commit()
+        printer = await printer_factory(model="X1C")
+
+        body = (
+            await _send(async_client, _manifest(action="queue", printer_id=printer.id, cost_center_id=424242))
+        ).json()
+
+        # Past the "is required" gate, so the id was passed on; refused only
+        # because this one does not exist.
+        assert body["queue_error"] in ("Cost center not found", "Estimated cost is required for cost center prints")
+
+    async def test_warns_when_the_named_printer_is_not_the_model_it_was_sliced_for(self, async_client, printer_factory):
+        printer = await printer_factory(model="H2D", name="Big one")
+        body = (await _send(async_client, _manifest(action="queue", printer_id=printer.id))).json()
+
+        assert body["queue"] is not None
+        assert any("Sliced for X1C" in w and "Big one" in w for w in body["warnings"])
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+class TestManifest:
+    async def test_an_unknown_manifest_version_is_refused_by_name(self, async_client):
+        manifest = _manifest()
+        manifest["schema_version"] = 2
+        response = await _send(async_client, manifest)
+        assert response.status_code == 400
+        assert "[1]" in response.json()["detail"]
+
+    async def test_malformed_manifest_is_a_422(self, async_client):
+        response = await async_client.post(
+            UPLOADS,
+            files={"file": ("a.gcode.3mf", _sliced_3mf(), "application/zip")},
+            data={"manifest": "{not json"},
+        )
+        assert response.status_code == 422
+
+    async def test_printer_and_target_model_together_are_refused(self, async_client):
+        response = await _send(async_client, _manifest(action="queue", printer_id=1, target_model="P1S"))
+        assert response.status_code == 422
+
+    async def test_unknown_keys_are_ignored_so_a_newer_plugin_still_works(self, async_client):
+        manifest = _manifest()
+        manifest["source"]["something_new"] = {"x": 1}
+        assert (await _send(async_client, manifest)).status_code == 200
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+class TestClaim:
+    async def test_a_failed_upload_releases_its_id_for_the_retry(self, async_client, test_engine):
+        manifest = _manifest()
+        bad = await _send(async_client, manifest, content=b"not a zip")
+        assert bad.status_code == 400
+        assert await _claims(test_engine) == []
+
+        good = await _send(async_client, manifest)
+        assert good.status_code == 200
+        assert good.json()["replayed"] is False
+
+    async def test_an_id_still_being_processed_is_refused(self, async_client, db_session):
+        upload_id = str(uuid.uuid4())
+        db_session.add(SlicerPluginUpload(upload_id=upload_id, caller="anonymous", status="processing"))
+        await db_session.commit()
+
+        response = await _send(async_client, _manifest(upload_id))
+        assert response.status_code == 409
+        assert "still being processed" in response.json()["detail"]
+
+    async def test_an_abandoned_claim_is_taken_over(self, async_client, db_session, test_engine):
+        upload_id = str(uuid.uuid4())
+        long_ago = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=1)
+        db_session.add(
+            SlicerPluginUpload(upload_id=upload_id, caller="anonymous", status="processing", created_at=long_ago)
+        )
+        await db_session.commit()
+
+        response = await _send(async_client, _manifest(upload_id))
+        assert response.status_code == 200, response.text
+        [claim] = await _claims(test_engine)
+        assert claim.status == "done"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+class TestPermissions:
+    async def test_info_without_auth(self, async_client):
+        body = (await async_client.get(INFO)).json()
+        assert body["manifest_versions"] == [1]
+        assert body["can_upload"] is True and body["can_queue"] is True
+
+    async def test_a_key_that_may_not_queue_is_told_so_and_refused_before_anything_is_stored(
+        self, async_client, db_session, printer_factory, test_engine
+    ):
+        key = await _admin_key(async_client, db_session, can_queue=False)
+        headers = {"X-API-Key": key}
+        printer = await printer_factory(model="X1C")
+
+        info = await async_client.get(INFO, headers=headers)
+        assert info.status_code == 200
+        assert info.json()["can_queue"] is False
+
+        refused = await _send(async_client, _manifest(action="queue", printer_id=printer.id), headers=headers)
+        assert refused.status_code == 403
+        assert await _count(test_engine, LibraryFile) == 0
+        assert await _claims(test_engine) == []
+
+        stored = await _send(async_client, _manifest(), headers=headers)
+        assert stored.status_code == 200
+
+    async def test_a_key_without_library_access_cannot_connect(self, async_client, db_session):
+        key = await _admin_key(async_client, db_session, can_manage_library=False)
+        assert (await async_client.get(INFO, headers={"X-API-Key": key})).status_code == 403
+
+    async def test_an_upload_id_does_not_replay_to_another_caller(self, async_client, db_session, test_engine):
+        key_a = await _admin_key(async_client, db_session)
+        full_b, hash_b, prefix_b = generate_api_key()
+        owner_id = (await db_session.execute(select(APIKey.user_id))).scalars().first()
+        db_session.add(
+            APIKey(
+                name="plugin-b",
+                key_hash=hash_b,
+                key_prefix=prefix_b,
+                enabled=True,
+                user_id=owner_id,
+                can_manage_library=True,
+                can_queue=True,
+            )
+        )
+        await db_session.commit()
+
+        manifest = _manifest()
+        assert (await _send(async_client, manifest, headers={"X-API-Key": key_a})).status_code == 200
+        other = await _send(async_client, manifest, headers={"Authorization": f"Bearer {full_b}"})
+
+        assert other.status_code == 409
+        assert "library_file" not in other.text
+        [claim] = await _claims(test_engine)
+        assert claim.caller.startswith("apikey:")