diff --git a/backend/app/api/routes/archives.py b/backend/app/api/routes/archives.py
index f5f902744..c4e6bc8a8 100644
--- a/backend/app/api/routes/archives.py
+++ b/backend/app/api/routes/archives.py
@@ -17,7 +17,9 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.core import database
 from backend.app.core.auth import (
+    ApiKeyActor,
     MediaOrRequestPrinterScope,
+    RequestActor,
     RequestPrinterScope,
     RequirePermissionIfAuthEnabled,
     probe_permissions_if_auth_enabled,
@@ -218,7 +220,7 @@ def _ensure_archive_visible(
     return archive
 
 
-def _validate_user_filter_permission(current_user: User | None, created_by_id: int | None):
+def _validate_user_filter_permission(current_user: User | ApiKeyActor | None, created_by_id: int | None):
     """Raise 403 if created_by_id filter is used without stats:filter_by_user permission."""
     if created_by_id is None or current_user is None:
         return
@@ -1139,9 +1141,10 @@ async def export_stats(
     db: AsyncSession = Depends(get_db),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.STATS_READ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Export statistics summary to CSV or Excel format."""
-    _validate_user_filter_permission(current_user, created_by_id)
+    _validate_user_filter_permission(actor, created_by_id)
 
     from fastapi.responses import StreamingResponse
 
@@ -1178,6 +1181,7 @@ async def get_archive_stats(
     db: AsyncSession = Depends(get_db),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.STATS_READ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Get statistics across all archives.
 
@@ -1188,7 +1192,7 @@ async def get_archive_stats(
     """
     from backend.app.models.print_log import PrintLogEntry
 
-    _validate_user_filter_permission(current_user, created_by_id)
+    _validate_user_filter_permission(actor, created_by_id)
 
     # Build date filter conditions scoped to PrintLogEntry (event-time).
     base_conditions = []
@@ -4915,6 +4919,7 @@ async def slice_archive(
     db: AsyncSession = Depends(get_db),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.LIBRARY_UPLOAD),
     printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Enqueue a slice job for an archive's source. Returns 202 + job_id;
     the slice runs in the background, the caller polls `GET /slice-jobs/{id}`.
@@ -4933,10 +4938,10 @@ async def slice_archive(
     archive = await db.get(PrintArchive, archive_id)
     # Per-row ownership gate — mirror the archive read routes. LIBRARY_UPLOAD
     # alone let a READ_OWN caller slice another user's archive by raw id even
-    # though GET on that id returned 404. API-key / auth-disabled callers
-    # (current_user is None) keep can_read_all=True — no per-row identity.
-    can_read_all = current_user is None or current_user.has_permission(Permission.ARCHIVES_READ_ALL.value)
-    archive = _ensure_archive_visible(archive, current_user, can_read_all, printer_scope)
+    # though GET on that id returned 404. 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.ARCHIVES_READ_ALL.value)
+    archive = _ensure_archive_visible(archive, actor, can_read_all, printer_scope)
 
     src_relative = archive.source_3mf_path or archive.file_path
     if not src_relative:
diff --git a/backend/app/api/routes/finance.py b/backend/app/api/routes/finance.py
index 0ace92fb3..87a55ae1a 100644
--- a/backend/app/api/routes/finance.py
+++ b/backend/app/api/routes/finance.py
@@ -7,7 +7,7 @@ from sqlalchemy import case, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
-from backend.app.core.auth import RequirePermissionIfAuthEnabled, require_auth_if_enabled
+from backend.app.core.auth import ApiKeyActor, RequestActor, RequirePermissionIfAuthEnabled, require_auth_if_enabled
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
 from backend.app.models.finance import (
@@ -638,8 +638,19 @@ async def create_manual_print(
 async def get_my_cost_centers(
     db: AsyncSession = Depends(get_db),
     current_user: User | None = Depends(require_auth_if_enabled),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
-    """Return private and assigned cost centers for the current user."""
+    """Return private and assigned cost centers for the current user.
+
+    An API key that may queue gets its owner's, the ones it can queue with
+    when billing is on (#3256). A key without an owner has none.
+    """
+    if isinstance(actor, ApiKeyActor):
+        if not actor.has_permission(Permission.QUEUE_CREATE.value):
+            raise HTTPException(status_code=403, detail="API key cannot queue prints")
+        if actor.owner is None:
+            return []
+        current_user = actor.owner
     user = await _require_authenticated_user(current_user)
 
     result = await db.execute(
diff --git a/backend/app/api/routes/library.py b/backend/app/api/routes/library.py
index b470e6032..6144874b5 100644
--- a/backend/app/api/routes/library.py
+++ b/backend/app/api/routes/library.py
@@ -25,7 +25,9 @@ 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
 from backend.app.core.auth import (
+    ApiKeyActor,
     QueueReviewRequired,
+    RequestActor,
     RequestPrinterScope,
     require_media_token_ownership,
     require_ownership_permission,
@@ -1240,11 +1242,12 @@ async def create_folder(
     data: FolderCreate,
     db: AsyncSession = Depends(get_db),
     current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """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, current_user)
+        await get_writable_folder(db, data.parent_id, actor)
 
     # Verify project exists if specified
     project_name = None
@@ -1269,14 +1272,14 @@ async def create_folder(
         parent_id=data.parent_id,
         project_id=data.project_id,
         archive_id=data.archive_id,
-        created_by_id=current_user.id if current_user else None,
-        # Made without a user (auth off, an API key): everyone's, as before #3201.
-        shared=current_user is None,
+        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, current_user)
+    index = await _load_index(db, actor)
 
     return FolderResponse(
         id=folder.id,
@@ -1294,7 +1297,7 @@ async def create_folder(
         # 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, current_user),
+        **_folder_access_fields(index, folder, actor),
         created_at=folder.created_at,
         updated_at=folder.updated_at,
     )
@@ -1874,6 +1877,7 @@ async def scan_external_folder(
     folder_id: int,
     db: AsyncSession = Depends(get_db),
     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.
 
@@ -1881,7 +1885,7 @@ async def scan_external_folder(
     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, current_user)
+    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")
 
@@ -2393,6 +2397,7 @@ async def upload_file(
     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)),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Upload a file to the library."""
     try:
@@ -2415,7 +2420,7 @@ async def upload_file(
         # 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, current_user)
+            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
@@ -2538,7 +2543,7 @@ async def upload_file(
             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,
+            created_by_id=actor.id if actor else None,
         )
         db.add(library_file)
         await db.commit()
@@ -2569,6 +2574,7 @@ async def extract_zip_file(
     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)),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Upload and extract a ZIP file to the library.
 
@@ -2586,7 +2592,7 @@ async def extract_zip_file(
 
     # 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, current_user)
+        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:
@@ -2635,7 +2641,7 @@ async def extract_zip_file(
         # 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, current_user)),
+            (f for f in existing.scalars().all() if can_write_folder(f, actor)),
             None,
         )
         if existing_folder:
@@ -2646,9 +2652,9 @@ async def extract_zip_file(
             new_folder = LibraryFolder(
                 name=zip_folder_name,
                 parent_id=folder_id,
-                created_by_id=current_user.id if current_user else None,
+                created_by_id=actor.id if actor else None,
                 # Made without a user (auth off, an API key): everyone's, as before #3201.
-                shared=current_user is None,
+                shared=actor is None or actor.id is None,
             )
             db.add(new_folder)
             await db.flush()
@@ -2700,7 +2706,7 @@ async def extract_zip_file(
                                         )
                                     )
                                     existing_folder = next(
-                                        (f for f in existing.scalars().all() if can_write_folder(f, current_user)),
+                                        (f for f in existing.scalars().all() if can_write_folder(f, actor)),
                                         None,
                                     )
 
@@ -2711,9 +2717,9 @@ async def extract_zip_file(
                                         new_folder = LibraryFolder(
                                             name=part,
                                             parent_id=current_parent,
-                                            created_by_id=current_user.id if current_user else None,
+                                            created_by_id=actor.id if actor else None,
                                             # Made without a user (auth off, an API key): everyone's, as before #3201.
-                                            shared=current_user is None,
+                                            shared=actor is None or actor.id is None,
                                         )
                                         db.add(new_folder)
                                         await db.flush()
@@ -2825,7 +2831,7 @@ async def extract_zip_file(
                         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,
+                        created_by_id=actor.id if actor else None,
                     )
                     db.add(library_file)
                     await db.flush()
@@ -3024,6 +3030,7 @@ async def combine_files(
     request: CombineFilesRequest,
     db: AsyncSession = Depends(get_db),
     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.
 
@@ -3044,11 +3051,11 @@ async def combine_files(
         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, current_user)
+        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 = current_user is None or current_user.has_permission(Permission.LIBRARY_READ_ALL.value)
+    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.
@@ -3061,7 +3068,7 @@ async def combine_files(
 
     # 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), current_user, can_read_all) for file_id in copies_by_id]
+    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:
@@ -3086,7 +3093,7 @@ async def combine_files(
         filename=filename,
         folder_id=request.folder_id,
         source_type="combined",
-        owner_id=current_user.id if current_user else None,
+        owner_id=actor.id if actor else None,
     )
 
     return FileUploadResponse(
@@ -3106,6 +3113,7 @@ async def add_files_to_queue(
     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.
 
@@ -3167,8 +3175,8 @@ async def add_files_to_queue(
     # 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 current_user is not None and not current_user.has_permission(Permission.LIBRARY_READ_ALL.value):
-        files = {fid: f for fid, f in files.items() if f.created_by_id == current_user.id}
+    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 == actor.id}
 
     # Project attribution (#1897): a file queued from a project-linked folder
     # inherits that project, so the resulting archive counts toward the
@@ -3286,7 +3294,7 @@ async def add_files_to_queue(
                 # 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=current_user.id if current_user else None,
+                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,
             )
@@ -5123,6 +5131,7 @@ async def slice_library_file(
     db: AsyncSession = Depends(get_db),
     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}`.
@@ -5139,10 +5148,10 @@ async def slice_library_file(
     # 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. API-key / auth-disabled callers
-    # (current_user is None) keep can_read_all=True — no per-row identity.
-    can_read_all = current_user is None or current_user.has_permission(Permission.LIBRARY_READ_ALL.value)
-    lib_file = _ensure_library_file_visible(lib_file, current_user, can_read_all)
+    # 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"):
diff --git a/backend/app/api/routes/makerworld.py b/backend/app/api/routes/makerworld.py
index ac1cd512d..7527369cc 100644
--- a/backend/app/api/routes/makerworld.py
+++ b/backend/app/api/routes/makerworld.py
@@ -32,6 +32,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from backend.app.api.routes.cloud import resolve_api_key_cloud_owner
 from backend.app.api.routes.library import save_3mf_bytes_to_library
 from backend.app.core.auth import (
+    ApiKeyActor,
+    RequestActor,
     RequirePermissionIfAuthEnabled,
     require_auth_if_enabled,
     require_permission_if_auth_enabled,
@@ -319,6 +321,7 @@ async def import_instance(
     credentials: HTTPAuthorizationCredentials | None = Depends(security),
     x_api_key: str | None = Header(default=None, alias="X-API-Key"),
     api_key_cloud_owner: User | None = Depends(resolve_api_key_cloud_owner),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Download a specific MakerWorld instance (plate configuration) and save
     the 3MF into the library.
@@ -338,7 +341,7 @@ async def import_instance(
 
     if body.folder_id is not None:
         # Only into a folder the user may write to (#3201).
-        target_folder = await get_writable_folder(db, body.folder_id, current_user)
+        target_folder = await get_writable_folder(db, body.folder_id, actor)
         if target_folder.is_external and target_folder.external_readonly:
             raise HTTPException(
                 status_code=403,
@@ -358,7 +361,7 @@ async def import_instance(
         if default_folder_name is None:
             effective_folder_id = None
         else:
-            default_folder = await default_import_folder(db, default_folder_name, current_user)
+            default_folder = await default_import_folder(db, default_folder_name, actor)
             effective_folder_id = default_folder.id
 
     service = await _build_service(db, provider, current_user, api_key_cloud_owner)
@@ -423,11 +426,8 @@ async def import_instance(
     # there as on the manifest-supplied name.
     filename = suggested_name if suggested_name.endswith(".3mf") else unquote(download.filename)
 
-    # API-keyed callers carry identity on the key, not in current_user (#1777);
-    # this collapse stays route-side solely so the library row is attributed
-    # to the key's owner rather than NULL. Credential identity is resolved
-    # inside the provider.
-    cloud_token_user = current_user or api_key_cloud_owner
+    # Credited to the key's owner for an API key. Credential identity is
+    # resolved inside the provider.
     library_file, was_existing = await save_3mf_bytes_to_library(
         db,
         file_bytes=download.file_bytes,
@@ -435,7 +435,7 @@ async def import_instance(
         folder_id=effective_folder_id,
         source_type=provider.source_type,
         source_url=source_url,
-        owner_id=cloud_token_user.id if cloud_token_user else None,
+        owner_id=actor.id if actor else None,
     )
 
     return MakerWorldImportResponse(
diff --git a/backend/app/api/routes/manyfold.py b/backend/app/api/routes/manyfold.py
index 46d152f3c..ade9fb946 100644
--- a/backend/app/api/routes/manyfold.py
+++ b/backend/app/api/routes/manyfold.py
@@ -21,7 +21,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.api.routes.library import save_3mf_bytes_to_library, validate_print_file_upload
 from backend.app.api.routes.settings import set_setting
-from backend.app.core.auth import RequirePermissionIfAuthEnabled
+from backend.app.core.auth import ApiKeyActor, RequestActor, RequirePermissionIfAuthEnabled
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
 from backend.app.models.library import LibraryFile
@@ -291,7 +291,7 @@ async def get_preview(
 # ---- import -------------------------------------------------------------
 
 
-async def _import_folder_id(db: AsyncSession, folder_id: int | None, user: User | None) -> int | None:
+async def _import_folder_id(db: AsyncSession, folder_id: int | None, user: User | ApiKeyActor | None) -> int | None:
     """The chosen folder, or the top-level "Manyfold" folder (created on first use)."""
     if folder_id is not None:
         # Only into a folder the user may write to (#3201).
@@ -308,6 +308,7 @@ async def import_file(
     body: ManyfoldImportRequest,
     db: AsyncSession = Depends(get_db),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.MANYFOLD_IMPORT),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Download one Manyfold file into the library.
 
@@ -343,7 +344,7 @@ async def import_file(
 
     filename = _library_filename(file["filename"], file_id)
     validate_print_file_upload(filename, data)
-    folder_id = await _import_folder_id(db, body.folder_id, current_user)
+    folder_id = await _import_folder_id(db, body.folder_id, actor)
     library_file, was_existing = await save_3mf_bytes_to_library(
         db,
         file_bytes=data,
@@ -351,7 +352,7 @@ async def import_file(
         folder_id=folder_id,
         source_type=manyfold_provider.source_type,
         source_url=source_url,
-        owner_id=current_user.id if current_user else None,
+        owner_id=actor.id if actor else None,
     )
     logger.info(
         "[MANYFOLD] Imported %s (model %s, file %s) as library file %s", filename, model_id, file_id, library_file.id
diff --git a/backend/app/api/routes/obico.py b/backend/app/api/routes/obico.py
index 69c43e653..232f2d283 100644
--- a/backend/app/api/routes/obico.py
+++ b/backend/app/api/routes/obico.py
@@ -5,7 +5,7 @@ import logging
 from fastapi import APIRouter, HTTPException, Response
 from pydantic import BaseModel
 
-from backend.app.core.auth import RequestPrinterScope, RequirePermissionIfAuthEnabled
+from backend.app.core.auth import ApiKeyActor, RequestActor, RequestPrinterScope, RequirePermissionIfAuthEnabled
 from backend.app.core.permissions import Permission
 from backend.app.core.printer_scope import PrinterScope
 from backend.app.models.user import User
@@ -44,6 +44,7 @@ async def get_status(
 async def get_printer_status(
     user: User | None = RequirePermissionIfAuthEnabled(Permission.PRINTERS_READ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Per-printer live classification for the printer cards (#1546).
 
@@ -54,7 +55,7 @@ async def get_printer_status(
     enabled_printers = settings["enabled_printers"]
     # Error strings can embed configured URLs (ML API base, external URL), so
     # they stay behind settings:read like the rest of the configuration.
-    can_see_error = user is None or user.has_permission(Permission.SETTINGS_READ.value)
+    can_see_error = actor is None or actor.has_permission(Permission.SETTINGS_READ.value)
     per_printer = obico_detection_service.get_per_printer()
     if not can_see_error:
         # The "error" *class* is not configuration — a printers:read user still
diff --git a/backend/app/api/routes/pipeline_runs.py b/backend/app/api/routes/pipeline_runs.py
index b28e977bb..78f7dbaad 100644
--- a/backend/app/api/routes/pipeline_runs.py
+++ b/backend/app/api/routes/pipeline_runs.py
@@ -34,7 +34,13 @@ from sqlalchemy import delete, desc, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.api.routes.cloud import resolve_api_key_cloud_owner
-from backend.app.core.auth import QueueReviewRequired, RequestPrinterScope, RequirePermissionIfAuthEnabled
+from backend.app.core.auth import (
+    ApiKeyActor,
+    QueueReviewRequired,
+    RequestActor,
+    RequestPrinterScope,
+    RequirePermissionIfAuthEnabled,
+)
 from backend.app.core.config import settings as app_settings
 from backend.app.core.database import async_session, get_db
 from backend.app.core.permissions import Permission
@@ -377,15 +383,15 @@ async def _resolve_source(
     *,
     library_file_id: int | None,
     archive_id: int | None,
-    user: User | None,
+    user: User | ApiKeyActor | None,
     printer_scope: PrinterScope,
 ) -> tuple[SourceKind, int, str, Path]:
     # Per-row ownership gate (IDOR fix): a caller may only run a pipeline on a
     # source they can see. Without this a READ_OWN caller could reference
     # another user's library file / archive by raw id and have it sliced (and,
     # via /run, printed) even though a direct GET on that id returned 404.
-    # Auth-disabled and API-key callers (user is None) keep can_read_all=True —
-    # no per-row identity, matching the library/archive read helpers.
+    # Only auth off (user is None) keeps can_read_all=True. An API key arrives
+    # as its owner's RequestActor, matching the library/archive read helpers.
     from backend.app.api.routes.archives import _ensure_archive_visible
     from backend.app.api.routes.library import _ensure_library_file_visible
 
@@ -653,6 +659,7 @@ async def check_eligibility(
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_READ),
     printer_scope: PrinterScope = RequestPrinterScope,
     db: AsyncSession = Depends(get_db),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     pipeline = await _load_pipeline(db, pipeline_id)
     printer_scope.ensure(pipeline.target_printer_id)
@@ -660,7 +667,7 @@ async def check_eligibility(
         db,
         library_file_id=body.source_library_file_id,
         archive_id=body.source_archive_id,
-        user=current_user,
+        user=actor,
         printer_scope=printer_scope,
     )
     if pipeline.target_kind == "printer_class" and pipeline.target_printer_id is None:
@@ -685,6 +692,7 @@ async def run_pipeline(
     printer_scope: PrinterScope = RequestPrinterScope,
     review_required: bool = QueueReviewRequired,
     db: AsyncSession = Depends(get_db),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     from backend.app.api.routes.settings import get_setting
     from backend.app.services.slice_dispatch import slice_dispatch
@@ -693,14 +701,13 @@ async def run_pipeline(
     # The pipeline is shared config; running it is limited to what the caller
     # may print on (#1727)
     printer_scope.ensure(pipeline.target_printer_id)
-    # ``user=current_user`` deliberately, not the cloud owner below: an API-key
-    # caller has no per-row identity and must keep can_read_all, the same as
-    # every other read helper.
+    # An API key is checked as its owner (RequestActor), like its owner's own
+    # session; ``creator`` below is a separate question.
     src_kind, src_id, src_filename, src_path = await _resolve_source(
         db,
         library_file_id=body.source_library_file_id,
         archive_id=body.source_archive_id,
-        user=current_user,
+        user=actor,
         printer_scope=printer_scope,
     )
 
@@ -993,6 +1000,7 @@ async def retry_failed(
     printer_scope: PrinterScope = RequestPrinterScope,
     review_required: bool = QueueReviewRequired,
     db: AsyncSession = Depends(get_db),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Create a new run with copies = (failed + cancelled count) from the
     parent. Same pipeline, same source. Eligibility re-checked at run time
@@ -1039,6 +1047,7 @@ async def retry_failed(
         body,
         current_user=current_user,
         api_key_cloud_owner=api_key_cloud_owner,
+        actor=actor,
         printer_scope=printer_scope,
         review_required=review_required,
         db=db,
diff --git a/backend/app/api/routes/print_queue.py b/backend/app/api/routes/print_queue.py
index f491ffb21..7a30453bf 100644
--- a/backend/app/api/routes/print_queue.py
+++ b/backend/app/api/routes/print_queue.py
@@ -15,7 +15,9 @@ from sqlalchemy.orm import selectinload
 
 from backend.app.api.routes.library_variants import normalize_model_name, resolve_variant_model
 from backend.app.core.auth import (
+    ApiKeyActor,
     QueueReviewRequired,
+    RequestActor,
     RequestPrinterScope,
     RequirePermissionIfAuthEnabled,
     RequirePrinterPermissionIfAuthEnabled,
@@ -167,7 +169,7 @@ def _extract_filament_types_from_3mf(file_path: Path, plate_id: int | None = Non
 _extract_print_time_from_3mf = extract_print_time_from_3mf
 
 
-def _assert_can_queue_archive(archive: PrintArchive, current_user: User | None) -> None:
+def _assert_can_queue_archive(archive: PrintArchive, current_user: User | ApiKeyActor | None) -> None:
     """Gate turning *archive* into a print. Raises rather than returning a verdict.
 
     Shared by every route that creates queue items from an archive, so a new
@@ -189,6 +191,7 @@ def _assert_can_queue_archive(archive: PrintArchive, current_user: User | None)
       allows any archive, REPRINT_OWN allows own only, ownerless archives
       require REPRINT_ALL (fail-closed).
     """
+    # None is auth off. An API key arrives as its RequestActor, never None.
     if current_user is None:
         return
     if not current_user.has_permission(Permission.ARCHIVES_READ_ALL.value) and archive.created_by_id != current_user.id:
@@ -204,8 +207,9 @@ def _assert_can_queue_archive(archive: PrintArchive, current_user: User | None)
         )
 
 
-def _assert_can_queue_library_file(library_file: LibraryFile, current_user: User | None) -> None:
+def _assert_can_queue_library_file(library_file: LibraryFile, current_user: User | ApiKeyActor | None) -> None:
     """Gate turning *library_file* into a print — LIBRARY_READ_ALL or ownership."""
+    # None is auth off. An API key arrives as its RequestActor, never None.
     if current_user is None:
         return
     if (
@@ -274,7 +278,9 @@ async def _is_orders_last_source(db: AsyncSession, item: PrintQueueItem) -> bool
     return survivor is None
 
 
-async def _assert_can_dispatch_batch_sources(db: AsyncSession, batch_id: int, current_user: User | None) -> None:
+async def _assert_can_dispatch_batch_sources(
+    db: AsyncSession, batch_id: int, current_user: User | ApiKeyActor | None
+) -> None:
     """Apply the ``POST /queue/`` source-file gates to everything a dispatch would print.
 
     Dispatching clones existing queue items, so without this it would be a
@@ -682,7 +688,7 @@ async def list_queue(
 async def _resolve_queue_variants(
     db: AsyncSession,
     specs: list[QueueVariantCreate],
-    current_user: User | None,
+    current_user: User | ApiKeyActor | None,
 ) -> list[tuple[QueueVariantCreate, LibraryFile, str]]:
     """Validate a cross-model candidate set and pair each file with its model (#671).
 
@@ -827,6 +833,7 @@ async def add_to_queue(
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_CREATE),
     printer_scope: PrinterScope = RequestPrinterScope,
     review_required: bool = QueueReviewRequired,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Add an item to the print queue."""
     # Normalize target_model (e.g., "Bambu Lab X1E" / "C13" -> "X1E").
@@ -851,7 +858,7 @@ async def add_to_queue(
             raise HTTPException(
                 400, "Cannot combine variants with archive_id or library_file_id — the variants are the files"
             )
-        variant_specs = await _resolve_queue_variants(db, data.variants, current_user)
+        variant_specs = await _resolve_queue_variants(db, data.variants, actor)
         # Mirror the first candidate onto the item so the queue listing, the SJF
         # grouping and the "Any H2S" label have something before a printer is
         # picked. Resolution overwrites it with whichever candidate actually runs.
@@ -869,6 +876,8 @@ async def add_to_queue(
     if data.printer_id and target_model_norm:
         raise HTTPException(400, "Cannot specify both printer_id and target_model")
     if target_model_norm:
+        # The key itself, not its actor: the scheduler holds an any-printer job
+        # to its creator's printers, and a key may be limited to fewer of them.
         ensure_model_target_allowed(current_user, printer_scope)
 
     # Validate printer exists (if assigned)
@@ -895,7 +904,7 @@ async def add_to_queue(
         archive = result.scalar_one_or_none()
         if not archive:
             raise HTTPException(400, "Archive not found")
-        _assert_can_queue_archive(archive, current_user)
+        _assert_can_queue_archive(archive, actor)
 
     # Validate library file exists (if provided) and get it for filament extraction
     library_file = None
@@ -904,7 +913,7 @@ async def add_to_queue(
         library_file = result.scalar_one_or_none()
         if not library_file:
             raise HTTPException(400, "Library file not found")
-        _assert_can_queue_library_file(library_file, current_user)
+        _assert_can_queue_library_file(library_file, actor)
         # Bambu SD card is FAT32/exFAT — illegal filename chars would 553 at
         # FTP upload time (#1540). Reject at queue time so the user gets the
         # actionable error before waiting in queue.
@@ -984,10 +993,10 @@ async def add_to_queue(
         if existing_batch.status != "active":
             raise HTTPException(400, "Cannot add items to a non-active batch")
         if (
-            current_user is not None
+            actor is not None
             and existing_batch.created_by_id is not None
-            and existing_batch.created_by_id != current_user.id
-            and not current_user.has_permission(Permission.QUEUE_UPDATE_ALL.value)
+            and existing_batch.created_by_id != actor.id
+            and not actor.has_permission(Permission.QUEUE_UPDATE_ALL.value)
         ):
             raise HTTPException(404, "Batch not found")
         batch = existing_batch
@@ -1021,7 +1030,7 @@ async def add_to_queue(
             library_file_id=data.library_file_id,
             quantity=quantity,
             status="active",
-            created_by_id=current_user.id if current_user else None,
+            created_by_id=actor.id if actor else None,
         )
         db.add(batch)
         await db.flush()  # Get batch.id before creating items
@@ -1106,7 +1115,7 @@ async def add_to_queue(
         db,
         cost_center_id=data.cost_center_id,
         estimated_cost=trusted_estimated_cost,
-        current_user=current_user,
+        current_user=actor,
         quantity=quantity,
     )
 
@@ -1195,7 +1204,7 @@ async def add_to_queue(
             project_id=data.project_id,
             position=start_position + i,
             status="pending",
-            created_by_id=current_user.id if current_user else None,
+            created_by_id=actor.id if actor else None,
             batch_id=batch_id,
             print_time_seconds=cached_print_time,
         )
@@ -1282,6 +1291,7 @@ async def bulk_update_queue_items(
         )
     ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Bulk update multiple queue items with the same values.
 
@@ -1351,7 +1361,7 @@ async def bulk_update_queue_items(
                 db,
                 cost_center_id=item_update_data.get("cost_center_id", item.cost_center_id),
                 estimated_cost=trusted_estimated_cost,
-                current_user=user,
+                current_user=actor,
                 exclude_queue_item_id=item.id,
             )
 
@@ -1400,7 +1410,9 @@ def _validate_plate_targets(
     return plates
 
 
-async def _validate_batch_project(db: AsyncSession, project_id: int | None, current_user: User | None) -> None:
+async def _validate_batch_project(
+    db: AsyncSession, project_id: int | None, current_user: User | ApiKeyActor | None
+) -> None:
     """404 on a bogus project id rather than letting the FK blow up as a 500."""
     if project_id is None:
         return
@@ -1410,7 +1422,7 @@ async def _validate_batch_project(db: AsyncSession, project_id: int | None, curr
 
 
 async def _load_batch_for_write(
-    db: AsyncSession, batch_id: int, current_user: User | None, permission: Permission
+    db: AsyncSession, batch_id: int, current_user: User | ApiKeyActor | None, permission: Permission
 ) -> PrintBatch:
     """Fetch a batch the caller is allowed to modify, or 404.
 
@@ -1459,6 +1471,7 @@ async def create_batch(
     db: AsyncSession = Depends(get_db),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_CREATE),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Create a batch.
 
@@ -1478,7 +1491,7 @@ async def create_batch(
         raise HTTPException(400, "Batch name is required")
 
     plate_targets = _validate_plate_targets(data.plates)
-    await _validate_batch_project(db, data.project_id, current_user)
+    await _validate_batch_project(db, data.project_id, actor)
     if data.external_source is not None:
         existing = await db.execute(
             select(PrintBatch.id).where(
@@ -1495,7 +1508,7 @@ async def create_batch(
         library_file_id=data.library_file_id,
         quantity=len(data.item_ids) if data.item_ids else 1,
         status="active",
-        created_by_id=current_user.id if current_user else None,
+        created_by_id=actor.id if actor else None,
         project_id=data.project_id,
         due_date=data.due_date,
         notes=data.notes,
@@ -1538,9 +1551,9 @@ async def create_batch(
             if not printer_scope.allows(item.printer_id):
                 continue
             if (
-                current_user is not None
-                and item.created_by_id != current_user.id
-                and not current_user.has_permission(Permission.QUEUE_UPDATE_ALL.value)
+                actor is not None
+                and item.created_by_id != actor.id
+                and not actor.has_permission(Permission.QUEUE_UPDATE_ALL.value)
             ):
                 continue
             item.batch_id = batch.id
@@ -1560,6 +1573,7 @@ async def update_batch(
     data: PrintBatchUpdate,
     db: AsyncSession = Depends(get_db),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_OWN),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Edit an order's header or its per-plate targets while it runs (#342).
 
@@ -1569,11 +1583,11 @@ async def update_batch(
     explicit action, because silently deleting queued work on a number change
     would be a nasty surprise.
     """
-    batch = await _load_batch_for_write(db, batch_id, current_user, Permission.QUEUE_UPDATE_ALL)
+    batch = await _load_batch_for_write(db, batch_id, actor, Permission.QUEUE_UPDATE_ALL)
 
     plate_targets = _validate_plate_targets(data.plates)
     if data.project_id is not None:
-        await _validate_batch_project(db, data.project_id, current_user)
+        await _validate_batch_project(db, data.project_id, actor)
 
     if data.name is not None:
         if not data.name.strip():
@@ -1632,6 +1646,7 @@ async def dispatch_batch(
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_CREATE),
     printer_scope: PrinterScope = RequestPrinterScope,
     review_required: bool = QueueReviewRequired,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Queue the runs this order still owes (#342).
 
@@ -1640,12 +1655,12 @@ async def dispatch_batch(
     overrides and print options the user already chose — and the validation
     those went through at creation time.
     """
-    batch = await _load_batch_for_write(db, batch_id, current_user, Permission.QUEUE_UPDATE_ALL)
+    batch = await _load_batch_for_write(db, batch_id, actor, Permission.QUEUE_UPDATE_ALL)
     if batch.status == "cancelled":
         raise HTTPException(400, "Cannot dispatch a cancelled batch")
 
     # Dispatch starts prints, so it must not be a weaker door than POST /queue/.
-    await _assert_can_dispatch_batch_sources(db, batch.id, current_user)
+    await _assert_can_dispatch_batch_sources(db, batch.id, actor)
 
     try:
         created = await dispatch_remaining(
@@ -1654,7 +1669,7 @@ async def dispatch_batch(
             plate_id=data.plate_id,
             only_plate=data.only_plate,
             limit=data.limit,
-            created_by_id=current_user.id if current_user else None,
+            created_by_id=actor.id if actor else None,
         )
     except BatchDispatchError as exc:
         raise HTTPException(400, str(exc)) from exc
@@ -1687,6 +1702,7 @@ async def ungroup_batch(
     db: AsyncSession = Depends(get_db),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_OWN),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Disband a batch: clear batch_id from all members and delete the batch row.
 
@@ -1698,8 +1714,8 @@ async def ungroup_batch(
     if not batch:
         raise HTTPException(404, "Batch not found")
 
-    can_modify_all = current_user is None or current_user.has_permission(Permission.QUEUE_UPDATE_ALL.value)
-    if not can_modify_all and batch.created_by_id != (current_user.id if current_user else None):
+    can_modify_all = actor is None or actor.has_permission(Permission.QUEUE_UPDATE_ALL.value)
+    if not can_modify_all and batch.created_by_id != (actor.id if actor else None):
         raise HTTPException(404, "Batch not found")
     await _ensure_batch_in_scope(db, batch_id, printer_scope)
 
@@ -1708,7 +1724,7 @@ async def ungroup_batch(
     ungrouped = 0
     remaining = 0
     for item in items:
-        if not can_modify_all and item.created_by_id != (current_user.id if current_user else None):
+        if not can_modify_all and item.created_by_id != (actor.id if actor else None):
             remaining += 1
             continue
         item.batch_id = None
@@ -1975,6 +1991,7 @@ async def update_queue_item(
         )
     ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Update a queue item."""
     user, can_modify_all = auth_result
@@ -2133,7 +2150,7 @@ async def update_queue_item(
         db,
         cost_center_id=update_data.get("cost_center_id", item.cost_center_id),
         estimated_cost=trusted_estimated_cost,
-        current_user=user,
+        current_user=actor,
         exclude_queue_item_id=item.id,
     )
 
@@ -2470,6 +2487,7 @@ async def start_queue_item(
         )
     ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Manually start a staged (manual_start) queue item.
 
@@ -2530,7 +2548,7 @@ async def start_queue_item(
         db,
         cost_center_id=item.cost_center_id,
         estimated_cost=item.estimated_cost,
-        current_user=user,
+        current_user=actor,
         exclude_queue_item_id=item.id,
     )
 
diff --git a/backend/app/api/routes/slicer_presets.py b/backend/app/api/routes/slicer_presets.py
index 51316846f..6facca1b1 100644
--- a/backend/app/api/routes/slicer_presets.py
+++ b/backend/app/api/routes/slicer_presets.py
@@ -26,7 +26,12 @@ from backend.app.api.routes.orca_cloud import (
     _build_authenticated_service as _build_orca_service,
     _load_credentials as _load_orca_credentials,
 )
-from backend.app.core.auth import RequestPrinterScope, RequirePermissionIfAuthEnabled, require_ownership_permission
+from backend.app.core.auth import (
+    RequestPrinterScope,
+    RequirePermissionIfAuthEnabled,
+    is_auth_enabled,
+    require_ownership_permission,
+)
 from backend.app.core.config import settings as app_settings
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
@@ -132,6 +137,10 @@ async def _fetch_cloud_presets(
     """
     if user is not None and not user.has_permission(Permission.CLOUD_AUTH.value):
         return _empty_slots(), "not_authenticated"
+    # The sign-in stored without a user is the auth-off install's, not one for
+    # a caller who has none while auth is on.
+    if user is None and await is_auth_enabled(db):
+        return _empty_slots(), "not_authenticated"
 
     token, _email, region = await get_stored_token(db, user)
     if not token:
@@ -212,6 +221,8 @@ async def _fetch_orca_cloud_presets(
     """
     if user is not None and not user.has_permission(Permission.ORCA_CLOUD_AUTH.value):
         return _empty_slots(), "not_authenticated"
+    if user is None and await is_auth_enabled(db):
+        return _empty_slots(), "not_authenticated"
 
     creds = await _load_orca_credentials(db, user)
     if not creds.token:
@@ -579,6 +590,7 @@ async def get_preset_values(
     slot: str = Query("process", description="Preset slot. Only 'process' is supported today."),
     db: AsyncSession = Depends(get_db),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.LIBRARY_UPLOAD),
+    api_key_cloud_owner: User | None = Depends(resolve_api_key_cloud_owner),
 ) -> dict:
     """Effective values of a preset, with its ``inherits:`` chain flattened.
 
@@ -610,7 +622,9 @@ async def get_preset_values(
         return {"resolved": False, "values": {}, "reason": reason}
 
     try:
-        profile_json = await resolve_preset_ref(db, current_user, ref, slot)
+        # A cloud preset resolves as the key's owner for a key with Allow
+        # Cloud Access, like the listing below.
+        profile_json = await resolve_preset_ref(db, current_user or api_key_cloud_owner, ref, slot)
     except HTTPException:
         # A preset the caller can't resolve is not a reason to break the panel;
         # the slice itself will report it properly if they go ahead.
diff --git a/backend/app/core/auth.py b/backend/app/core/auth.py
index e23eec5b9..91c211d29 100644
--- a/backend/app/core/auth.py
+++ b/backend/app/core/auth.py
@@ -116,12 +116,9 @@ _APIKEY_SCOPE_BY_PERMISSION: dict[Permission, str | tuple[str, ...]] = {
     Permission.PRINTER_SENSOR_HISTORY_READ: "can_read_status",
     Permission.STATS_READ: "can_read_status",
     Permission.STATS_FILTER_BY_USER: "can_read_status",
-    # USERS_READ_SLIM grants no data an API key could not already reach (#1894):
-    # for API-keyed requests the permission deps return None as ``current_user``,
-    # so ``_validate_user_filter_permission`` in routes/archives.py short-circuits
-    # and ``?created_by_id=N`` is already honoured for every N. Without a way to
-    # discover the ids, that filter is only addressable by brute force. The slim
-    # listing makes it usable; the full USERS_READ listing (emails, roles, group
+    # USERS_READ_SLIM is ids and usernames only (#1894). It lets a key whose
+    # owner holds stats:filter_by_user address ``?created_by_id=N`` without
+    # guessing ids. The full USERS_READ listing (emails, roles, group
     # membership, permission sets) stays unmapped = admin-only.
     Permission.USERS_READ_SLIM: "can_read_status",
     Permission.SYSTEM_READ: "can_read_status",
@@ -491,6 +488,38 @@ def _check_apikey_permissions(
         raise last_failure
 
 
+class ApiKeyActor:
+    """An API key standing in for its owner in a route's own per-row checks.
+
+    Permission dependencies answer a key request with no user, and a route's
+    own checks read no user as "auth is off": they skip the archive, library
+    and cost-center ownership tests a signed-in session faces. Routes that
+    make those tests take this from ``RequestActor`` instead. It holds only
+    the permissions the key may exercise (``apikey_effective_permissions``),
+    so it never exceeds the owner nor the key's scope flags, and it is an
+    administrator, for checks such as printing with any cost center, only when
+    the owner is one. A legacy key without an owner has no ``id``, so it owns
+    nothing and is a member of no cost center.
+    """
+
+    def __init__(self, api_key: APIKey, owner: User | None):
+        self.api_key = api_key
+        self.owner = owner
+        self.is_admin: bool = owner is not None and owner.is_admin
+        self.id: int | None = owner.id if owner is not None else None
+        self.username: str | None = owner.username if owner is not None else None
+        self._permissions = frozenset(apikey_effective_permissions(api_key, owner))
+
+    def has_permission(self, permission: str) -> bool:
+        return permission in self._permissions
+
+    def has_all_permissions(self, *permissions: str) -> bool:
+        return all(p in self._permissions for p in permissions)
+
+    def has_any_permission(self, *permissions: str) -> bool:
+        return any(p in self._permissions for p in permissions)
+
+
 @dataclass(frozen=True)
 class ScopedCaller:
     """Who passed a scoped door: an API key, a user, or nobody (auth disabled)."""
@@ -2012,6 +2041,41 @@ async def get_queue_review_required(
 QueueReviewRequired = Depends(get_queue_review_required)
 
 
+async def get_request_actor(
+    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(security)] = None,
+    x_api_key: Annotated[str | None, Header(alias="X-API-Key")] = None,
+) -> User | ApiKeyActor | None:
+    """FastAPI dependency: who a route's own ownership checks run against.
+
+    The signed-in user, or for an API key an ``ApiKeyActor`` for its owner.
+    None only when auth is off. Declare it after the permission dependency,
+    which has already turned away bad credentials.
+    """
+    async with async_session() as db:
+        if not await is_auth_enabled(db):
+            return None
+        api_key = await validated_api_key_from_request(credentials, x_api_key)
+        if api_key is not None:
+            return ApiKeyActor(api_key, await resolve_apikey_owner(db, api_key))
+    user = None
+    if credentials is not None:
+        cached = _authenticated_user.get()
+        if cached is not None and cached[0] == credentials.credentials:
+            user = cached[1]
+        else:
+            user = await get_current_user_optional(credentials)
+    if user is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Authentication required",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    return user
+
+
+RequestActor = Depends(get_request_actor)
+
+
 async def get_media_or_request_printer_scope(
     token: str | None = None,
     credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(security)] = None,
diff --git a/backend/app/services/library_folder_access.py b/backend/app/services/library_folder_access.py
index 32a752159..71daa9880 100644
--- a/backend/app/services/library_folder_access.py
+++ b/backend/app/services/library_folder_access.py
@@ -2,7 +2,8 @@
 
 A folder has an owner (``created_by_id``, the user who made it) and can be
 marked ``shared`` by an admin. A user with ``library:read_all`` (or any
-caller when auth is off, or an API key) sees every folder. A user with only
+caller when auth is off) sees every folder. An API key is passed in as its
+``ApiKeyActor`` and treated as its owner within its scopes. A user with only
 ``library:read_own`` sees:
 
   - folders they own,
@@ -14,7 +15,7 @@ They may write into (upload, extract, move files, create subfolders in)
 their own folders and shared ones, plus the root. Everything else is hidden:
 a folder they can't see answers 404, exactly like another user's file.
 
-A folder made without a user (auth off, an API key) is created shared, so
+A folder made without a user (auth off, a key without an owner) is created shared, so
 switching auth on later doesn't hide it. Folders from before #3201 got an
 owner or the shared flag from the upgrade backfill in ``core/database.py``.
 """
@@ -33,7 +34,7 @@ from backend.app.models.user import User
 
 
 def sees_all_folders(user: User | None) -> bool:
-    """True for ``library:read_all``, and for ``None`` (auth off or an API key)."""
+    """True for ``library:read_all``, and for ``None`` (auth off)."""
     return user is None or user.has_permission(Permission.LIBRARY_READ_ALL.value)
 
 
diff --git a/backend/app/services/model_providers/makerworld/provider.py b/backend/app/services/model_providers/makerworld/provider.py
index 9393c3dde..a1afd7c90 100644
--- a/backend/app/services/model_providers/makerworld/provider.py
+++ b/backend/app/services/model_providers/makerworld/provider.py
@@ -14,6 +14,7 @@ from typing import TYPE_CHECKING
 
 import httpx
 
+from backend.app.core.auth import is_auth_enabled
 from backend.app.core.permissions import Permission
 from backend.app.services.model_providers.base import (
     ModelProvider,
@@ -78,7 +79,12 @@ class MakerWorldProvider(ModelProvider):
         flag those installs read back on the status endpoints.
         """
         identity = user if user is not None else api_key_owner
-        token, _email, _region = await get_stored_token(db, identity)
+        # Without an identity, the stored sign-in is the auth-off install's.
+        # With auth on (an API key without Allow Cloud Access) there is none.
+        if identity is None and await is_auth_enabled(db):
+            token = None
+        else:
+            token, _email, _region = await get_stored_token(db, identity)
         user_id = identity.id if identity is not None else None
         return MakerWorldService(
             client=client,
diff --git a/backend/app/services/preset_resolver.py b/backend/app/services/preset_resolver.py
index dd8607730..868493c9f 100644
--- a/backend/app/services/preset_resolver.py
+++ b/backend/app/services/preset_resolver.py
@@ -30,6 +30,7 @@ from fastapi import HTTPException
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.api.routes.orca_cloud import _build_authenticated_service as _build_orca_service
+from backend.app.core.auth import is_auth_enabled
 from backend.app.core.permissions import Permission
 from backend.app.models.local_preset import LocalPreset
 from backend.app.models.user import User
@@ -111,6 +112,12 @@ async def _resolve_local(db: AsyncSession, ref: PresetRef, slot: str) -> str:
     return preset.setting
 
 
+# The sign-in stored without a user belongs to an install with auth off. With
+# auth on, a caller without a user (an API key without Allow Cloud Access)
+# has no cloud sign-in of its own and must not borrow that one.
+_NO_CLOUD_IDENTITY = "{cloud} presets need a signed-in user or an API key with Allow Cloud Access ({slot})"
+
+
 async def _resolve_cloud(db: AsyncSession, user: User | None, ref: PresetRef, slot: str) -> str:
     """Fetch a single cloud preset detail. Permission gate matches the
     rest of the cloud surface (`CLOUD_AUTH`) so a user with `LIBRARY_UPLOAD`
@@ -121,6 +128,8 @@ async def _resolve_cloud(db: AsyncSession, user: User | None, ref: PresetRef, sl
             status_code=403,
             detail=f"Cloud presets require the cloud:auth permission ({slot})",
         )
+    if user is None and await is_auth_enabled(db):
+        raise HTTPException(status_code=403, detail=_NO_CLOUD_IDENTITY.format(cloud="Bambu Cloud", slot=slot))
 
     token, _email, region = await get_stored_token(db, user)
     if not token:
@@ -198,6 +207,8 @@ async def _resolve_orca_cloud(db: AsyncSession, user: User | None, ref: PresetRe
             status_code=403,
             detail=f"Orca Cloud presets require the orca_cloud:auth permission ({slot})",
         )
+    if user is None and await is_auth_enabled(db):
+        raise HTTPException(status_code=403, detail=_NO_CLOUD_IDENTITY.format(cloud="Orca Cloud", slot=slot))
 
     try:
         svc = await _build_orca_service(db, user)
diff --git a/backend/tests/integration/test_api_key_cloud_access.py b/backend/tests/integration/test_api_key_cloud_access.py
index 567e9eb56..5844a7e15 100644
--- a/backend/tests/integration/test_api_key_cloud_access.py
+++ b/backend/tests/integration/test_api_key_cloud_access.py
@@ -414,3 +414,57 @@ class TestSliceRouteCloudOwnerResolution:
             db=db_session,
         )
         assert owner is None
+
+
+class TestPresetValuesResolvesAsCloudOwner:
+    """GET /slicer/preset-values resolves a cloud preset as the key's
+    owner when the key has Allow Cloud Access, like the preset listing does.
+    A key without it has no cloud identity there."""
+
+    async def _resolved_as(self, client: AsyncClient, db: AsyncSession, *, can_access_cloud: bool):
+        from unittest.mock import AsyncMock, patch
+
+        from fastapi import HTTPException
+
+        await _setup_auth_with_admin(client)
+        owner = await _store_admin_cloud_token(db, "cloudadmin", token="fake-token")
+        full_key, key_hash, key_prefix = generate_api_key()
+        db.add(
+            APIKey(
+                name="presets",
+                key_hash=key_hash,
+                key_prefix=key_prefix,
+                user_id=owner.id,
+                can_manage_library=True,
+                can_access_cloud=can_access_cloud,
+            )
+        )
+        await db.commit()
+
+        resolve = AsyncMock(side_effect=HTTPException(status_code=400, detail="stop here"))
+        with patch("backend.app.api.routes.slicer_presets.resolve_preset_ref", resolve):
+            resp = await client.get(
+                "/api/v1/slicer/preset-values",
+                params={"source": "cloud", "id": "PFUS123", "slot": "process"},
+                headers={"X-API-Key": full_key},
+            )
+        assert resp.status_code == 200, resp.text
+        assert resp.json()["resolved"] is False
+        user = resolve.await_args.args[1]
+        return owner, user
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_key_with_cloud_scope_resolves_as_its_owner(
+        self, async_client: AsyncClient, db_session: AsyncSession
+    ):
+        owner, user = await self._resolved_as(async_client, db_session, can_access_cloud=True)
+        assert user is not None and user.id == owner.id
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_key_without_cloud_scope_has_no_cloud_identity(
+        self, async_client: AsyncClient, db_session: AsyncSession
+    ):
+        _owner, user = await self._resolved_as(async_client, db_session, can_access_cloud=False)
+        assert user is None
diff --git a/backend/tests/integration/test_api_key_queue_acts_as_owner.py b/backend/tests/integration/test_api_key_queue_acts_as_owner.py
new file mode 100644
index 000000000..6264f109d
--- /dev/null
+++ b/backend/tests/integration/test_api_key_queue_acts_as_owner.py
@@ -0,0 +1,473 @@
+"""An API key acts as its owner (#3256).
+
+A route's own checks on cost centers, archives, library files and folders run
+against the key's owner, with the owner's permissions narrowed to the key's
+scope flags. Where it applies, each case goes through the key and through the
+owner's session, which must agree. A key made before keys had owners owns
+nothing and may use no cost center.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from backend.app.core.auth import generate_api_key, get_password_hash
+from backend.app.core.config import settings as app_settings
+from backend.app.models.api_key import APIKey
+from backend.app.models.archive import PrintArchive
+from backend.app.models.finance import CostCenter, CostCenterMember
+from backend.app.models.group import Group
+from backend.app.models.library import LibraryFile, LibraryFolder
+from backend.app.models.print_batch import PrintBatch
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+
+PASSWORD = "Ownerpass1!"
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+
+async def _set(db_session, key: str, value: str) -> None:
+    row = await db_session.scalar(select(Settings).where(Settings.key == key))
+    if row is None:
+        db_session.add(Settings(key=key, value=value))
+    else:
+        row.value = value
+
+
+@pytest.fixture
+async def world(db_session):
+    """Auth and billing on; a key owner who may queue and reprint only their
+    own archives, another user, and one printer."""
+    await _set(db_session, "auth_enabled", "true")
+    await _set(db_session, "advanced_auth_enabled", "false")
+    await _set(db_session, "billing_enabled", "true")
+    group = Group(
+        name="own-queuers",
+        description="t",
+        permissions=[
+            "queue:create",
+            "queue:read_own",
+            "queue:update_own",
+            "archives:read_own",
+            "archives:reprint_own",
+            "library:read_own",
+        ],
+        is_system=False,
+    )
+    db_session.add(group)
+    await db_session.flush()
+    owner = User(username="keyowner", password_hash=get_password_hash(PASSWORD), is_active=True, groups=[group])
+    other = User(username="otheruser", password_hash=get_password_hash(PASSWORD), is_active=True)
+    admin = User(username="adminowner", password_hash=get_password_hash(PASSWORD), role="admin", is_active=True)
+    printer = Printer(
+        name="P", ip_address="192.168.9.9", serial_number="00M00A3256000001", access_code="12345678", model="X1C"
+    )
+    db_session.add_all([owner, other, admin, printer])
+    await db_session.commit()
+    return {"owner": owner, "other": other, "admin": admin, "printer": printer}
+
+
+async def _archive(db_session, created_by_id: int | None, n: int) -> PrintArchive:
+    archive = PrintArchive(
+        filename=f"a{n}.3mf",
+        print_name=f"a{n}",
+        file_path=f"/tmp/a3256_{n}.3mf",  # nosec B108
+        file_size=1,
+        content_hash=f"hash3256_{n}",
+        status="completed",
+        cost=1.25,
+        filament_used_grams=50.0,
+        created_by_id=created_by_id,
+    )
+    db_session.add(archive)
+    await db_session.commit()
+    await db_session.refresh(archive)
+    return archive
+
+
+async def _center(db_session, *, owner_user_id: int | None = None, member_id: int | None = None) -> CostCenter:
+    center = CostCenter(
+        name=f"cc-{owner_user_id}-{member_id}",
+        is_active=True,
+        is_private=owner_user_id is not None,
+        owner_user_id=owner_user_id,
+    )
+    db_session.add(center)
+    await db_session.flush()
+    if member_id is not None:
+        db_session.add(CostCenterMember(cost_center_id=center.id, user_id=member_id, can_print=True))
+    await db_session.commit()
+    await db_session.refresh(center)
+    return center
+
+
+async def _key(db_session, owner_id: int | None, *, can_queue: bool = True) -> dict[str, str]:
+    full_key, key_hash, key_prefix = generate_api_key()
+    db_session.add(
+        APIKey(
+            name="probe",
+            key_hash=key_hash,
+            key_prefix=key_prefix,
+            user_id=owner_id,
+            can_queue=can_queue,
+            can_read_status=True,
+        )
+    )
+    await db_session.commit()
+    return {"X-API-Key": full_key}
+
+
+async def _login(client: AsyncClient, username: str) -> dict[str, str]:
+    response = await client.post("/api/v1/auth/login", json={"username": username, "password": PASSWORD})
+    assert response.status_code == 200, response.text
+    return {"Authorization": f"Bearer {response.json()['access_token']}"}
+
+
+async def _queue(client: AsyncClient, headers, printer: Printer, archive: PrintArchive, cost_center_id: int | None):
+    return await client.post(
+        "/api/v1/queue/",
+        json={"printer_id": printer.id, "archive_id": archive.id, "cost_center_id": cost_center_id},
+        headers=headers,
+    )
+
+
+class TestCostCenters:
+    async def test_another_users_private_cost_center_is_refused(self, async_client, db_session, world):
+        archive = await _archive(db_session, world["owner"].id, 1)
+        center = await _center(db_session, owner_user_id=world["other"].id)
+        key = await _key(db_session, world["owner"].id)
+
+        by_key = await _queue(async_client, key, world["printer"], archive, center.id)
+        by_session = await _queue(
+            async_client, await _login(async_client, "keyowner"), world["printer"], archive, center.id
+        )
+
+        assert by_key.status_code == by_session.status_code == 403
+        assert await db_session.scalar(select(PrintQueueItem)) is None
+
+    async def test_a_shared_cost_center_needs_membership(self, async_client, db_session, world):
+        archive = await _archive(db_session, world["owner"].id, 1)
+        center = await _center(db_session, member_id=world["other"].id)
+        key = await _key(db_session, world["owner"].id)
+
+        assert (await _queue(async_client, key, world["printer"], archive, center.id)).status_code == 403
+
+    async def test_the_owners_own_cost_center_works_and_the_item_is_the_owners(self, async_client, db_session, world):
+        archive = await _archive(db_session, world["owner"].id, 1)
+        center = await _center(db_session, owner_user_id=world["owner"].id)
+        key = await _key(db_session, world["owner"].id)
+
+        response = await _queue(async_client, key, world["printer"], archive, center.id)
+
+        assert response.status_code == 200, response.text
+        item = await db_session.scalar(select(PrintQueueItem).where(PrintQueueItem.id == response.json()["id"]))
+        assert item.cost_center_id == center.id
+        # Credited to the owner, so the scheduler's check at print start has
+        # someone to check, and the owner sees it under queue:read_own.
+        assert item.created_by_id == world["owner"].id
+
+    async def test_a_cost_center_the_owner_is_a_member_of_works(self, async_client, db_session, world):
+        archive = await _archive(db_session, world["owner"].id, 1)
+        center = await _center(db_session, member_id=world["owner"].id)
+        key = await _key(db_session, world["owner"].id)
+
+        assert (await _queue(async_client, key, world["printer"], archive, center.id)).status_code == 200
+
+    async def test_an_admins_key_may_use_any_cost_center_like_the_admin(self, async_client, db_session, world):
+        archive = await _archive(db_session, world["admin"].id, 1)
+        center = await _center(db_session, owner_user_id=world["other"].id)
+        key = await _key(db_session, world["admin"].id)
+
+        assert (await _queue(async_client, key, world["printer"], archive, center.id)).status_code == 200
+
+    async def test_a_key_without_an_owner_may_use_no_cost_center(self, async_client, db_session, world):
+        archive = await _archive(db_session, None, 1)
+        center = await _center(db_session, member_id=world["owner"].id)
+        key = await _key(db_session, None)
+
+        assert (await _queue(async_client, key, world["printer"], archive, center.id)).status_code == 403
+
+    async def test_moving_an_item_to_a_forbidden_cost_center_is_refused(self, async_client, db_session, world):
+        # PATCH through a key needs the owner to hold queue:update_all.
+        group = await db_session.scalar(select(Group).where(Group.name == "own-queuers"))
+        group.permissions = [*group.permissions, "queue:update_all"]
+        await db_session.commit()
+        archive = await _archive(db_session, world["owner"].id, 1)
+        allowed = await _center(db_session, owner_user_id=world["owner"].id)
+        forbidden = await _center(db_session, owner_user_id=world["other"].id)
+        key = await _key(db_session, world["owner"].id)
+        created = await _queue(async_client, key, world["printer"], archive, allowed.id)
+        assert created.status_code == 200, created.text
+
+        response = await async_client.patch(
+            f"/api/v1/queue/{created.json()['id']}", json={"cost_center_id": forbidden.id}, headers=key
+        )
+
+        assert response.status_code == 403
+        item = await db_session.scalar(select(PrintQueueItem).where(PrintQueueItem.id == created.json()["id"]))
+        await db_session.refresh(item)
+        assert item.cost_center_id == allowed.id
+
+
+class TestListingCostCenters:
+    async def test_a_key_lists_its_owners_cost_centers(self, async_client, db_session, world):
+        mine = await _center(db_session, owner_user_id=world["owner"].id)
+        shared = await _center(db_session, member_id=world["owner"].id)
+        await _center(db_session, owner_user_id=world["other"].id)
+        key = await _key(db_session, world["owner"].id)
+
+        by_key = await async_client.get("/api/v1/finance/cost-centers/mine", headers=key)
+        by_session = await async_client.get(
+            "/api/v1/finance/cost-centers/mine", headers=await _login(async_client, "keyowner")
+        )
+
+        assert by_key.status_code == 200, by_key.text
+        assert {c["id"] for c in by_key.json()} == {mine.id, shared.id}
+        assert by_key.json() == by_session.json()
+
+    async def test_a_key_that_cannot_queue_gets_none(self, async_client, db_session, world):
+        await _center(db_session, owner_user_id=world["owner"].id)
+        key = await _key(db_session, world["owner"].id, can_queue=False)
+
+        assert (await async_client.get("/api/v1/finance/cost-centers/mine", headers=key)).status_code == 403
+
+    async def test_a_key_without_an_owner_has_none(self, async_client, db_session, world):
+        await _center(db_session, owner_user_id=world["owner"].id)
+        key = await _key(db_session, None)
+
+        response = await async_client.get("/api/v1/finance/cost-centers/mine", headers=key)
+
+        assert response.status_code == 200
+        assert response.json() == []
+
+
+class TestSources:
+    async def test_another_users_archive_is_not_found(self, async_client, db_session, world):
+        await _set(db_session, "billing_enabled", "false")
+        archive = await _archive(db_session, world["other"].id, 1)
+        key = await _key(db_session, world["owner"].id)
+
+        by_key = await _queue(async_client, key, world["printer"], archive, None)
+        by_session = await _queue(async_client, await _login(async_client, "keyowner"), world["printer"], archive, None)
+
+        assert by_key.status_code == by_session.status_code == 404
+
+    async def test_another_users_library_file_is_not_found(self, async_client, db_session, world):
+        await _set(db_session, "billing_enabled", "false")
+        rel_path = "archive/library/files/probe_3256.gcode.3mf"
+        abs_path = Path(app_settings.base_dir) / rel_path
+        abs_path.parent.mkdir(parents=True, exist_ok=True)
+        abs_path.write_bytes(b"probe")
+        library_file = LibraryFile(
+            filename="probe_3256.gcode.3mf",
+            file_path=rel_path,
+            file_size=5,
+            file_type="3mf",
+            created_by_id=world["other"].id,
+        )
+        db_session.add(library_file)
+        await db_session.commit()
+        key = await _key(db_session, world["owner"].id)
+        try:
+            response = await async_client.post(
+                "/api/v1/queue/",
+                json={"printer_id": world["printer"].id, "library_file_id": library_file.id},
+                headers=key,
+            )
+        finally:
+            abs_path.unlink(missing_ok=True)
+
+        assert response.status_code == 404
+
+    async def test_the_owners_own_archive_still_queues(self, async_client, db_session, world):
+        await _set(db_session, "billing_enabled", "false")
+        archive = await _archive(db_session, world["owner"].id, 1)
+        key = await _key(db_session, world["owner"].id)
+
+        assert (await _queue(async_client, key, world["printer"], archive, None)).status_code == 200
+
+
+class TestBatches:
+    async def test_another_users_batch_cannot_be_changed(self, async_client, db_session, world):
+        await _set(db_session, "billing_enabled", "false")
+        batch = PrintBatch(name="theirs", created_by_id=world["other"].id)
+        db_session.add(batch)
+        await db_session.commit()
+        key = await _key(db_session, world["owner"].id)
+
+        update = await async_client.patch(f"/api/v1/queue/batches/{batch.id}", json={"name": "mine"}, headers=key)
+        ungroup = await async_client.post(f"/api/v1/queue/batches/{batch.id}/ungroup", headers=key)
+
+        assert update.status_code == 404
+        assert ungroup.status_code in (403, 404)
+        await db_session.refresh(batch)
+        assert batch.name == "theirs"
+
+
+@pytest.fixture
+async def library_world(db_session, world):
+    """The same owner, who may also upload, read stats and pipelines, and a
+    key with the library and status scopes as well."""
+    await _set(db_session, "billing_enabled", "false")
+    group = await db_session.scalar(select(Group).where(Group.name == "own-queuers"))
+    group.permissions = [*group.permissions, "library:upload", "stats:read", "pipelines:read"]
+    full_key, key_hash, key_prefix = generate_api_key()
+    db_session.add(
+        APIKey(
+            name="library probe",
+            key_hash=key_hash,
+            key_prefix=key_prefix,
+            user_id=world["owner"].id,
+            can_queue=True,
+            can_read_status=True,
+            can_manage_library=True,
+        )
+    )
+    theirs = LibraryFolder(name="theirs", created_by_id=world["other"].id, shared=False)
+    db_session.add(theirs)
+    await db_session.commit()
+    return {**world, "key": {"X-API-Key": full_key}, "their_folder": theirs}
+
+
+async def _their_file(db_session, world, filename: str = "theirs.stl") -> LibraryFile:
+    row = LibraryFile(
+        filename=filename,
+        file_path=f"library/files/{filename}",
+        file_type=filename.rsplit(".", 1)[-1],
+        file_size=1024,
+        created_by_id=world["other"].id,
+    )
+    db_session.add(row)
+    await db_session.commit()
+    await db_session.refresh(row)
+    return row
+
+
+_SLICE_BODY = {"printer_preset_id": 1, "process_preset_id": 2, "filament_preset_id": 3}
+
+
+class TestLibrary:
+    async def test_no_folder_inside_another_users_private_folder(self, async_client, db_session, library_world):
+        body = {"name": "sneaky", "parent_id": library_world["their_folder"].id}
+
+        by_key = await async_client.post("/api/v1/library/folders", json=body, headers=library_world["key"])
+        by_session = await async_client.post(
+            "/api/v1/library/folders", json=body, headers=await _login(async_client, "keyowner")
+        )
+
+        assert by_key.status_code == by_session.status_code == 404
+
+    async def test_a_keys_folder_is_its_owners_and_not_shared(self, async_client, db_session, library_world):
+        response = await async_client.post(
+            "/api/v1/library/folders", json={"name": "mine"}, headers=library_world["key"]
+        )
+
+        assert response.status_code == 200, response.text
+        folder = await db_session.get(LibraryFolder, response.json()["id"])
+        assert folder.created_by_id == library_world["owner"].id
+        assert folder.shared is False
+
+    async def test_no_upload_into_another_users_private_folder(self, async_client, db_session, library_world):
+        response = await async_client.post(
+            "/api/v1/library/files",
+            params={"folder_id": library_world["their_folder"].id},
+            files={"file": ("cube.stl", b"solid cube\nendsolid cube\n", "application/octet-stream")},
+            headers=library_world["key"],
+        )
+
+        assert response.status_code == 404
+
+    async def test_no_combining_another_users_file(self, async_client, db_session, library_world):
+        theirs = await _their_file(db_session, library_world)
+
+        response = await async_client.post(
+            "/api/v1/library/files/combine",
+            json={"items": [{"file_id": theirs.id}], "filename": "mix"},
+            headers=library_world["key"],
+        )
+
+        assert response.status_code == 404
+        assert response.json()["detail"] == "File not found"
+
+    async def test_no_slicing_another_users_file(self, async_client, db_session, library_world):
+        theirs = await _their_file(db_session, library_world)
+
+        response = await async_client.post(
+            f"/api/v1/library/files/{theirs.id}/slice", json=_SLICE_BODY, headers=library_world["key"]
+        )
+
+        assert response.status_code == 404
+        assert response.json()["detail"] == "File not found"
+
+    async def test_no_queueing_another_users_file_from_the_library(self, async_client, db_session, library_world):
+        theirs = await _their_file(db_session, library_world, "theirs.gcode.3mf")
+
+        response = await async_client.post(
+            "/api/v1/library/files/add-to-queue",
+            json={"file_ids": [theirs.id], "printer_id": library_world["printer"].id},
+            headers=library_world["key"],
+        )
+
+        # The same answer an unknown id gets
+        assert response.status_code == 400
+        assert response.json()["detail"]["errors"][0]["error"] == "File not found"
+        assert await db_session.scalar(select(PrintQueueItem)) is None
+
+
+class TestArchivesAndPipelines:
+    async def test_no_slicing_another_users_archive(self, async_client, db_session, library_world):
+        archive = await _archive(db_session, library_world["other"].id, 1)
+
+        response = await async_client.post(
+            f"/api/v1/archives/{archive.id}/slice", json=_SLICE_BODY, headers=library_world["key"]
+        )
+
+        assert response.status_code == 404
+        assert response.json()["detail"] == "Archive not found"
+
+    async def test_no_per_user_stats_without_the_owners_permission(self, async_client, db_session, library_world):
+        by_key = await async_client.get(
+            "/api/v1/archives/stats",
+            params={"created_by_id": library_world["other"].id},
+            headers=library_world["key"],
+        )
+        by_session = await async_client.get(
+            "/api/v1/archives/stats",
+            params={"created_by_id": library_world["other"].id},
+            headers=await _login(async_client, "keyowner"),
+        )
+
+        assert by_key.status_code == by_session.status_code == 403
+
+    async def test_no_pipeline_on_another_users_file(self, async_client, db_session, library_world):
+        theirs = await _their_file(db_session, library_world)
+        created = await async_client.post(
+            "/api/v1/slicer-pipelines/",
+            json={
+                "name": "Batch",
+                "description": None,
+                "printer_preset": {"source": "local", "id": "1"},
+                "process_preset": {"source": "local", "id": "2"},
+                "filament_presets": [{"source": "local", "id": "3"}],
+                "bed_type": None,
+            },
+            headers=await _login(async_client, "adminowner"),
+        )
+        assert created.status_code == 201, created.text
+
+        response = await async_client.post(
+            f"/api/v1/slicer-pipelines/{created.json()['id']}/check-eligibility",
+            json={"source_library_file_id": theirs.id},
+            headers=library_world["key"],
+        )
+
+        # Refused by the ownership check, not later for the missing file
+        assert response.status_code == 404
+        assert response.json()["detail"] == "File not found"
diff --git a/backend/tests/integration/test_makerworld_apikey_auth.py b/backend/tests/integration/test_makerworld_apikey_auth.py
index 0b1da7915..0845d5be0 100644
--- a/backend/tests/integration/test_makerworld_apikey_auth.py
+++ b/backend/tests/integration/test_makerworld_apikey_auth.py
@@ -254,7 +254,7 @@ class TestImportEndpoint:
 
     @pytest.mark.asyncio
     @pytest.mark.integration
-    async def test_api_key_without_cloud_scope_still_imports_but_owner_is_none(
+    async def test_api_key_without_cloud_scope_imports_anonymously_for_its_owner(
         self, async_client: AsyncClient, db_session: AsyncSession
     ):
         """Fail-closed parity: a key with can_manage_library but NOT
@@ -262,9 +262,9 @@ class TestImportEndpoint:
         the cloud-token resolver returns None, so the service is built
         without a token. The MakerWorldService itself would 401 on
         get_profile_download in production — here we just confirm the
-        route doesn't suddenly grant cloud identity from a non-cloud key,
-        and that the library row's owner_id stays NULL when there's no
-        resolved cloud-scoped owner.
+        route doesn't suddenly grant cloud identity from a non-cloud key.
+        The library row still belongs to the key's owner: crediting the file
+        is not a cloud question (#3256).
         """
         await _setup_auth_with_admin(async_client)
         admin = await _store_admin_cloud_token(db_session, "mwadmin", token="fake-bambu-token")
@@ -298,10 +298,9 @@ class TestImportEndpoint:
         assert jwt_user is None
         assert key_owner is None
 
-        # And owner_id is NULL because the cloud-scope fence said no.
         result = await db_session.execute(select(LibraryFile).where(LibraryFile.id == body["library_file_id"]))
         saved = result.scalar_one()
-        assert saved.created_by_id is None
+        assert saved.created_by_id == admin.id
 
 
 class TestJwtPathUnchanged:
diff --git a/backend/tests/integration/test_obico_api.py b/backend/tests/integration/test_obico_api.py
index 314c618f4..aab7cc042 100644
--- a/backend/tests/integration/test_obico_api.py
+++ b/backend/tests/integration/test_obico_api.py
@@ -203,7 +203,7 @@ class TestObicoPrinterStatusNoVerdict:
         # redaction under test is independent of them.
         loaded = {"enabled": True, "enabled_printers": None}
         with patch.object(obico_detection_service, "_load_settings", new=AsyncMock(return_value=loaded)):
-            data = await get_printer_status(user=user, printer_scope=ALL_PRINTERS)
+            data = await get_printer_status(user=user, printer_scope=ALL_PRINTERS, actor=user)
         entry = data["per_printer"][1]
         assert entry["class"] == "error"
         assert entry["error"] is None
diff --git a/backend/tests/unit/services/test_model_provider_interface.py b/backend/tests/unit/services/test_model_provider_interface.py
index 03406c0d6..51e0ff4e5 100644
--- a/backend/tests/unit/services/test_model_provider_interface.py
+++ b/backend/tests/unit/services/test_model_provider_interface.py
@@ -112,6 +112,35 @@ class TestBuildService:
     read the caller's stored Bambu Cloud token and wire the rejected-token
     callback so a 401 invalidates the shared credential app-wide."""
 
+    @pytest.fixture(autouse=True)
+    def _auth_off(self):
+        """These describe the auth-off install, the only one whose sign-in is
+        stored without a user (see the auth-on test below)."""
+        with patch(
+            "backend.app.services.model_providers.makerworld.provider.is_auth_enabled",
+            AsyncMock(return_value=False),
+        ):
+            yield
+
+    @pytest.mark.asyncio
+    async def test_with_auth_on_no_identity_borrows_no_stored_sign_in(self):
+        """With auth on, a caller without a user (an API key without Allow
+        Cloud Access) gets no token: the sign-in stored without a user is the
+        auth-off install's, and may be left over after auth was turned on."""
+        stored = AsyncMock(return_value=("global-tok", "admin@x.com", "global"))
+        with (
+            patch(
+                "backend.app.services.model_providers.makerworld.provider.is_auth_enabled",
+                AsyncMock(return_value=True),
+            ),
+            patch("backend.app.services.model_providers.makerworld.provider.get_stored_token", stored),
+        ):
+            svc = await makerworld_provider.build_service(db=AsyncMock(), user=None)
+
+        assert svc._auth_token is None
+        stored.assert_not_awaited()
+        await svc.close()
+
     @pytest.mark.asyncio
     async def test_seeds_token_and_auth_failure_callback(self):
         db = AsyncMock()
diff --git a/backend/tests/unit/services/test_preset_resolver.py b/backend/tests/unit/services/test_preset_resolver.py
index bd0bf3a83..84d08a2a9 100644
--- a/backend/tests/unit/services/test_preset_resolver.py
+++ b/backend/tests/unit/services/test_preset_resolver.py
@@ -435,3 +435,30 @@ async def test_resolve_preset_ref_dispatches_by_source():
         db, user, PresetRef(source="standard", id="Some Bundled Name"), slot="printer"
     )
     assert json.loads(out)["inherits"] == "Some Bundled Name"
+
+
+# --- no user while auth is on ----------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "resolve,source,loader",
+    [
+        (preset_resolver._resolve_cloud, "cloud", "get_stored_token"),
+        (preset_resolver._resolve_orca_cloud, "orca_cloud", "_build_orca_service"),
+    ],
+)
+async def test_no_user_with_auth_on_borrows_no_stored_sign_in(resolve, source, loader):
+    """The sign-in stored without a user is the auth-off install's, and may be
+    left over after auth was turned on. A caller with no user while auth is
+    on (an API key without Allow Cloud Access) must not slice with it."""
+    load = AsyncMock(return_value=("global-tok", "admin@x.com", "global"))
+    with (
+        patch.object(preset_resolver, "is_auth_enabled", AsyncMock(return_value=True)),
+        patch.object(preset_resolver, loader, load),
+        pytest.raises(HTTPException) as exc,
+    ):
+        await resolve(MagicMock(), None, PresetRef(source=source, id="X"), slot="printer")
+
+    assert exc.value.status_code == 403
+    load.assert_not_awaited()
diff --git a/backend/tests/unit/test_slicer_presets.py b/backend/tests/unit/test_slicer_presets.py
index dd9195107..b8b2cb5cc 100644
--- a/backend/tests/unit/test_slicer_presets.py
+++ b/backend/tests/unit/test_slicer_presets.py
@@ -904,3 +904,33 @@ class TestListPrinterModels:
 
         result = sp.list_printer_models()
         assert result is not PRINTER_MODEL_MAP
+
+
+class TestNoUserWithAuthOn:
+    """The sign-in stored without a user is the auth-off install's, and may be
+    left over after auth was turned on. A caller with no user while auth is
+    on (an API key without Allow Cloud Access) must not list its presets."""
+
+    @pytest.mark.asyncio
+    async def test_bambu_cloud(self):
+        sp._cloud_cache.clear()
+        with (
+            patch.object(sp, "is_auth_enabled", AsyncMock(return_value=True)),
+            patch.object(sp, "get_stored_token", AsyncMock(return_value=("global-tok", None, None))) as get_tok,
+        ):
+            slots, status = await sp._fetch_cloud_presets(MagicMock(), None)
+        assert status == "not_authenticated"
+        assert slots == {"printer": [], "process": [], "filament": []}
+        get_tok.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_orca_cloud(self):
+        sp._orca_cloud_cache.clear()
+        with (
+            patch.object(sp, "is_auth_enabled", AsyncMock(return_value=True)),
+            patch.object(sp, "_load_orca_credentials", AsyncMock()) as load,
+        ):
+            slots, status = await sp._fetch_orca_cloud_presets(MagicMock(), None)
+        assert status == "not_authenticated"
+        assert slots == {"printer": [], "process": [], "filament": []}
+        load.assert_not_called()