Security hardening (maziggy/bambuddy-security #20)

.basedpyright/baseline.json

@@ -918,14 +918,6 @@
                     "endColumn": 102,
                     "lineCount": 1
                 }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 26,
-                    "endColumn": 32,
-                    "lineCount": 1
-                }
             }
         ],
         "./backend/app/api/routes/print_log.py": [

backend/app/api/routes/archives.py

@@ -5171,6 +5171,9 @@ async def slice_archive(
     model_bytes = src_path.read_bytes()
     archive_id_local = archive.id
     user_id = current_user.id if current_user else None
+    # The new archive is its owner's also when a key, which resolves no cloud
+    # presets, slices it.
+    owner_id = user_id if user_id is not None else (actor.id if isinstance(actor, ApiKeyActor) else None)
 
     # Block a cross-nozzle-class re-slice (single-nozzle <-> H2D) up front —
     # BambuStudio's multi-extruder validator would otherwise reject it with a
@@ -5194,6 +5197,7 @@ async def slice_archive(
                     source_archive=src_archive,
                     current_user_id=user_id,
                     job_id=job_id,
+                    owner_id=owner_id,
                 )
             except HTTPException as exc:
                 raise http_exception_to_job_error(exc) from exc
@@ -5203,7 +5207,7 @@ async def slice_archive(
         kind="archive",
         source_id=archive.id,
         source_name=archive.print_name or archive.filename or f"archive {archive.id}",
-        owner_id=user_id,
+        owner_id=owner_id,
         run=_run,
     )
     return {

backend/app/api/routes/bulk.py

@@ -33,6 +33,7 @@ from backend.app.api.routes.smart_plugs import (
 )
 from backend.app.core.auth import (
     ApiKeyActor,
+    RequestActor,
     RequestPrinterScope,
     RequirePermissionIfAuthEnabled,
     require_ownership_permission,
@@ -260,6 +261,7 @@ async def get_folders_by_archives(
         )
     ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """``GET /library/folders/by-archive/{archive_id}`` for several archives, keyed by archive id.
 
@@ -270,4 +272,4 @@ async def get_folders_by_archives(
     archive_ids = parse_id_list(ids)
     if not archive_ids:
         return {}
-    return await _folders_by_archive(db, user, archive_ids, printer_scope)
+    return await _folders_by_archive(db, user, archive_ids, printer_scope, actor)

backend/app/api/routes/library.py

@@ -39,6 +39,7 @@ from backend.app.core.config import settings as app_settings
 from backend.app.core.database import async_session, get_db
 from backend.app.core.permissions import Permission
 from backend.app.core.printer_scope import PrinterScope, ensure_model_target_allowed
+from backend.app.core.row_visibility import visible_row_clauses
 from backend.app.core.tasks import spawn_background_task
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile, LibraryFileTag, LibraryFolder
@@ -163,18 +164,24 @@ def _ensure_library_file_visible(
     return library_file
 
 
-async def _linkable_archives(db: AsyncSession, archive_ids, printer_scope: PrinterScope) -> dict[int, str | None]:
-    """Print names of the archives among ``archive_ids`` on the caller's printers.
+async def _linkable_archives(
+    db: AsyncSession, archive_ids, printer_scope: PrinterScope, user: User | ApiKeyActor | None
+) -> dict[int, str | None]:
+    """Print names of the archives among ``archive_ids`` the caller sees in the archives.
 
-    A folder's link to any other archive reads as no link, and linking one
-    answers as for a missing archive.
+    Live archives on the caller's printers, their own unless they see
+    everyone's. A folder's link to any other archive reads as no link, and
+    linking one answers as for a missing archive.
     """
     ids = {i for i in archive_ids if i}
-    if not ids:
+    visible = visible_row_clauses(
+        PrintArchive, printer_scope, user, Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN
+    )
+    if not ids or visible is None:
         return {}
-    query = select(PrintArchive.id, PrintArchive.print_name).where(PrintArchive.id.in_(ids))
-    if (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
-        query = query.where(clause)
+    query = select(PrintArchive.id, PrintArchive.print_name).where(
+        PrintArchive.id.in_(ids), PrintArchive.deleted_at.is_(None), *visible
+    )
     return dict((await db.execute(query)).all())
 
 
@@ -688,6 +695,7 @@ async def save_3mf_bytes_to_library(
     source_type: str | None = None,
     source_url: str | None = None,
     owner_id: int | None = None,
+    dedupe_within: list | tuple | None = (),
 ) -> tuple[LibraryFile, bool]:
     """Save a fetched file into the library and return ``(library_file, was_existing)``.
 
@@ -697,14 +705,18 @@ async def save_3mf_bytes_to_library(
     if a LibraryFile with the same source_url already exists, the existing
     row is returned and the bytes are NOT re-saved (MakerWorld signed URLs
     change each download, so hash-based dedupe alone would miss re-imports).
+    ``dedupe_within`` narrows the rows that count as that copy to the ones the
+    importer may open; None imports a copy of their own regardless.
 
     Parses 3MF metadata + thumbnail the same way the multipart upload route
     does, via :class:`ThreeMFParser`, and renders an STL's thumbnail. Paths are stored as relative so the
     library is portable across installs.
     """
     # Source-URL-based dedupe: return the existing row untouched.
-    if source_url:
-        existing = await db.execute(LibraryFile.active().where(LibraryFile.source_url == source_url).limit(1))
+    if source_url and dedupe_within is not None:
+        existing = await db.execute(
+            LibraryFile.active().where(LibraryFile.source_url == source_url, *dedupe_within).limit(1)
+        )
         existing_row = existing.scalar_one_or_none()
         if existing_row is not None:
             return existing_row, True
@@ -722,6 +734,26 @@ async def save_3mf_bytes_to_library(
         folder_q = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
         target_folder = folder_q.scalar_one_or_none()
 
+    # Another user's copy of this very import, which the importer may not
+    # open, may already sit on the mount under this name: theirs is stored
+    # beside it under a free name, as a slice is. Any other file of that name
+    # is still refused, so nothing put there outside Bambuddy is touched.
+    if (
+        source_url
+        and target_folder is not None
+        and target_folder.is_external
+        and target_folder.external_path
+        and Path(target_folder.external_path).is_dir()
+        and await db.scalar(
+            select(LibraryFile.id)
+            .where(LibraryFile.source_url == source_url, LibraryFile.deleted_at.is_(None))
+            .limit(1)
+        )
+        is not None
+    ):
+        with contextlib.suppress(PathTraversalError):
+            filename = _unique_external_name(Path(target_folder.external_path), filename)
+
     file_path, is_external = _resolve_upload_destination(target_folder, filename)
     ext = file_path.suffix.lower() or ".3mf"
     with open(file_path, "wb") as fh:
@@ -1029,6 +1061,7 @@ async def list_folders(
         )
     ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Get all folders as a tree structure.
 
@@ -1049,7 +1082,7 @@ async def list_folders(
         .order_by(LibraryFolder.name)
     )
     rows = [row for row in result.all() if visible is None or row[0].id in visible]
-    archives = await _linkable_archives(db, [row[0].archive_id for row in rows], printer_scope)
+    archives = await _linkable_archives(db, [row[0].archive_id for row in rows], printer_scope, actor)
     own_files = _own_files_filter(user)
 
     # Get file counts per folder
@@ -1149,6 +1182,7 @@ async def get_folders_by_project(
         )
     ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Get all folders linked to a specific project, those the user may see (#3201)."""
     user, _ = auth_result
@@ -1162,7 +1196,7 @@ async def get_folders_by_project(
         .order_by(LibraryFolder.name)
     )
     rows = [row for row in result.all() if visible is None or row[0].id in visible]
-    archives = await _linkable_archives(db, [row[0].archive_id for row in rows], printer_scope)
+    archives = await _linkable_archives(db, [row[0].archive_id for row in rows], printer_scope, actor)
 
     folders = []
     for folder, project_name in rows:
@@ -1217,21 +1251,27 @@ async def get_folders_by_archive(
         )
     ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Get all folders linked to a specific archive, those the user may see (#3201)."""
     user, _ = auth_result
-    return (await _folders_by_archive(db, user, [archive_id], printer_scope))[archive_id]
+    return (await _folders_by_archive(db, user, [archive_id], printer_scope, actor))[archive_id]
 
 
 async def _folders_by_archive(
-    db: AsyncSession, user: User | ApiKeyActor | None, archive_ids: list[int], printer_scope: PrinterScope
+    db: AsyncSession,
+    user: User | ApiKeyActor | None,
+    archive_ids: list[int],
+    printer_scope: PrinterScope,
+    actor: User | ApiKeyActor | None,
 ) -> dict[int, list]:
     """The folders linked to each archive that *user* may see (#3201), by archive id.
 
-    An archive on a printer outside *printer_scope* has none.
+    An archive *actor* (the caller, as ``RequestActor`` gives it) does not see
+    in the archives has none.
     """
     folders: dict[int, list] = {archive_id: [] for archive_id in archive_ids}
-    archives = await _linkable_archives(db, archive_ids, printer_scope)
+    archives = await _linkable_archives(db, archive_ids, printer_scope, actor)
     if not archives:
         return folders
     index = await _load_index(db, user)
@@ -1308,7 +1348,7 @@ async def create_folder(
         project_name = project.name
 
     # Verify archive exists if specified
-    archives = await _linkable_archives(db, [data.archive_id], printer_scope)
+    archives = await _linkable_archives(db, [data.archive_id], printer_scope, actor)
     if data.archive_id is not None and data.archive_id not in archives:
         raise HTTPException(status_code=404, detail="Archive not found")
 
@@ -1358,6 +1398,7 @@ async def get_folder(
         )
     ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Get a folder by ID; 404 for a folder the user may not see (#3201)."""
     user, _ = auth_result
@@ -1376,7 +1417,7 @@ async def get_folder(
         raise HTTPException(status_code=404, detail="Folder not found")
 
     folder, project_name = row
-    archives = await _linkable_archives(db, [folder.archive_id], printer_scope)
+    archives = await _linkable_archives(db, [folder.archive_id], printer_scope, actor)
 
     # Get file count + latest file activity (#1770) in one trip
     agg_result = await db.execute(
@@ -1488,6 +1529,7 @@ async def update_folder(
         )
     ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Update a folder.
 
@@ -1548,11 +1590,11 @@ async def update_folder(
     if data.archive_id is not None:
         if data.archive_id == 0:
             # A link the caller can't see is not theirs to remove.
-            if folder.archive_id in await _linkable_archives(db, [folder.archive_id], printer_scope):
+            if folder.archive_id in await _linkable_archives(db, [folder.archive_id], printer_scope, actor):
                 folder.archive_id = None
         else:
             # Verify archive exists
-            if data.archive_id not in await _linkable_archives(db, [data.archive_id], printer_scope):
+            if data.archive_id not in await _linkable_archives(db, [data.archive_id], printer_scope, actor):
                 raise HTTPException(status_code=404, detail="Archive not found")
             folder.archive_id = data.archive_id
 
@@ -1579,7 +1621,7 @@ async def update_folder(
     if folder.project_id:
         project_result = await db.execute(select(Project.name).where(Project.id == folder.project_id))
         project_name = project_result.scalar()
-    archives = await _linkable_archives(db, [folder.archive_id], printer_scope)
+    archives = await _linkable_archives(db, [folder.archive_id], printer_scope, actor)
 
     return FolderResponse(
         id=folder.id,
@@ -4850,6 +4892,10 @@ async def guard_nozzle_class_reslice(
     return None
 
 
+# slice_and_persist's default owner: the user it slices for.
+_CALLER = object()
+
+
 async def slice_and_persist(
     db: AsyncSession,
     *,
@@ -4860,10 +4906,14 @@ async def slice_and_persist(
     request: SliceRequest,
     current_user_id: int | None,
     job_id: int | None = None,
+    owner_id: int | None | object = _CALLER,
 ) -> SliceResponse:
     """Slice a model and save the result as a new ``LibraryFile`` in
     ``folder_id`` (same folder as the source by convention).
 
+    ``current_user_id`` resolves cloud presets; the new file is ``owner_id``'s,
+    by default the same user.
+
     Always exports as ``.gcode.3mf`` so the existing library thumbnail
     pipeline works on the new file. Plain ``.gcode`` would have no
     embedded thumbnail to extract.
@@ -4965,7 +5015,7 @@ async def slice_and_persist(
     from backend.app.services.slice_dispatch import slice_dispatch as _dispatch
 
     # Nobody's, when the user was deleted while it sliced: the id may be someone else's now.
-    current_user_id = _dispatch.owner_now(job_id, current_user_id)
+    file_owner_id = _dispatch.owner_now(job_id, current_user_id if owner_id is _CALLER else owner_id)  # type: ignore[arg-type]
     new_file = LibraryFile(
         folder_id=folder_id,
         is_external=out_is_external,
@@ -4983,7 +5033,7 @@ async def slice_and_persist(
         thumbnail_path=thumbnail_relative,
         file_metadata=metadata,
         source_type="sliced",
-        created_by_id=current_user_id,
+        created_by_id=file_owner_id,
     )
     db.add(new_file)
     await db.commit()
@@ -5011,6 +5061,7 @@ async def slice_and_persist_as_archive(
     source_archive,  # PrintArchive — hint kept loose to avoid cyclic import
     current_user_id: int | None,
     job_id: int | None = None,
+    owner_id: int | None | object = _CALLER,
 ):
     """Slice a model and save the result as a new ``PrintArchive`` row,
     inheriting printer / project / makerworld metadata from the source
@@ -5154,7 +5205,7 @@ async def slice_and_persist_as_archive(
     from backend.app.services.slice_dispatch import slice_dispatch as _dispatch
 
     # Nobody's, when the user was deleted while it sliced: the id may be someone else's now.
-    current_user_id = _dispatch.owner_now(job_id, current_user_id)
+    current_user_id = _dispatch.owner_now(job_id, current_user_id if owner_id is _CALLER else owner_id)  # type: ignore[arg-type]
     new_archive = PrintArchive(
         printer_id=new_printer_id,
         project_id=source_archive.project_id,
@@ -5275,6 +5326,9 @@ async def slice_library_file(
     # cloud_token (#1182 follow-up).
     cloud_token_user = current_user or api_key_cloud_owner
     user_id = cloud_token_user.id if cloud_token_user else None
+    # The sliced file is its owner's also when no one resolves cloud presets,
+    # as a pipeline run's is.
+    owner_id = user_id if user_id is not None else (actor.id if isinstance(actor, ApiKeyActor) else None)
 
     # If the source has a `print_name` in its metadata (BambuStudio always
     # sets this; OrcaSlicer often leaves it blank), derive the sliced
@@ -5312,6 +5366,7 @@ async def slice_library_file(
                     request=request,
                     current_user_id=user_id,
                     job_id=job_id,
+                    owner_id=owner_id,
                 )
             except HTTPException as exc:
                 raise http_exception_to_job_error(exc) from exc
@@ -5321,7 +5376,7 @@ async def slice_library_file(
         kind="library_file",
         source_id=lib_file.id,
         source_name=lib_file.filename,
-        owner_id=user_id,
+        owner_id=owner_id,
         run=_run,
     )
     return {

backend/app/api/routes/makerworld.py

@@ -42,6 +42,7 @@ from backend.app.core.auth import (
 )
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
+from backend.app.core.row_visibility import owned_row_clauses
 from backend.app.models.library import LibraryFile
 from backend.app.models.user import User
 from backend.app.schemas.makerworld import (
@@ -260,6 +261,7 @@ async def resolve_url(
     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,
 ):
     """Resolve a MakerWorld URL to full model metadata + plate list.
 
@@ -292,13 +294,18 @@ async def resolve_url(
     # the provider's :meth:`source_url_filter` owns what "belongs" means
     # (whole-model key, per-plate keys, ...). The frontend surfaces the ids
     # to mark imported plates in the instance picker.
-    existing_q = await db.execute(
-        select(LibraryFile.id).where(
-            provider.source_url_filter(LibraryFile.source_url, str(model_id)),
-            LibraryFile.deleted_at.is_(None),
+    # Only rows the caller may open in the File Manager.
+    readable = owned_row_clauses(LibraryFile, actor, Permission.LIBRARY_READ_ALL, Permission.LIBRARY_READ_OWN)
+    already_imported = []
+    if readable is not None:
+        existing_q = await db.execute(
+            select(LibraryFile.id).where(
+                provider.source_url_filter(LibraryFile.source_url, str(model_id)),
+                LibraryFile.deleted_at.is_(None),
+                *readable,
+            )
         )
-    )
-    already_imported = [row[0] for row in existing_q.all()]
+        already_imported = [row[0] for row in existing_q.all()]
 
     return MakerWorldResolvedModel(
         model_id=model_id,
@@ -388,9 +395,15 @@ async def import_instance(
         # entry (see ``ModelProvider.canonical_url``).
         source_url = provider.canonical_url(info.ref)
 
-        # Dedupe check upfront so we don't burn bandwidth re-downloading.
-        existing_q = await db.execute(LibraryFile.active().where(LibraryFile.source_url == source_url).limit(1))
-        existing_row = existing_q.scalar_one_or_none()
+        # Dedupe check upfront so we don't burn bandwidth re-downloading. A copy
+        # the caller cannot open is no copy for them: they get their own.
+        readable = owned_row_clauses(LibraryFile, actor, Permission.LIBRARY_READ_ALL, Permission.LIBRARY_READ_OWN)
+        existing_row = None
+        if readable is not None:
+            existing_q = await db.execute(
+                LibraryFile.active().where(LibraryFile.source_url == source_url, *readable).limit(1)
+            )
+            existing_row = existing_q.scalar_one_or_none()
         if existing_row is not None:
             return MakerWorldImportResponse(
                 library_file_id=existing_row.id,
@@ -439,6 +452,7 @@ async def import_instance(
         source_type=provider.source_type,
         source_url=source_url,
         owner_id=actor.id if actor else None,
+        dedupe_within=readable,
     )
 
     return MakerWorldImportResponse(
@@ -455,8 +469,9 @@ async def recent_imports(
     limit: int = 10,
     db: AsyncSession = Depends(get_db, scope="function"),
     current_user: User | None = RequirePermissionIfAuthEnabled(makerworld_provider.view_permission),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
-    """Last N MakerWorld imports, newest first.
+    """Last N MakerWorld imports the caller may open, newest first.
 
     Surfaces files whose ``source_type`` is ``"makerworld"`` so the MakerWorld
     page can show a 'Recent imports' sidebar that persists across resolves.
@@ -467,9 +482,12 @@ async def recent_imports(
     _ = current_user  # permission gate only
     capped = max(1, min(50, int(limit)))
 
+    readable = owned_row_clauses(LibraryFile, actor, Permission.LIBRARY_READ_ALL, Permission.LIBRARY_READ_OWN)
+    if readable is None:
+        return []
     result = await db.execute(
         LibraryFile.active()
-        .where(LibraryFile.source_type == makerworld_provider.source_type)
+        .where(LibraryFile.source_type == makerworld_provider.source_type, *readable)
         .order_by(LibraryFile.created_at.desc())
         .limit(capped)
     )

backend/app/api/routes/manyfold.py

@@ -24,6 +24,7 @@ from backend.app.api.routes.settings import set_setting
 from backend.app.core.auth import ApiKeyActor, RequestActor, RequirePermissionIfAuthEnabled, require_actor_still_exists
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
+from backend.app.core.row_visibility import owned_row_clauses
 from backend.app.models.library import LibraryFile
 from backend.app.models.settings import Settings
 from backend.app.models.user import User
@@ -229,8 +230,10 @@ async def get_model(
     model_id: str,
     db: AsyncSession = Depends(get_db, scope="function"),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.MANYFOLD_VIEW),
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
-    """A model's details and files, with the library file of each earlier import."""
+    """A model's details and files, with the library file of each earlier import
+    the caller may open."""
     model_id = _id(model_id)
     service = await _service(db)
     try:
@@ -241,12 +244,17 @@ async def get_model(
         await service.close()
 
     imported: dict[str, ManyfoldLibraryRef] = {}
-    rows = await db.execute(
-        LibraryFile.active()
-        .where(manyfold_provider.source_url_filter(LibraryFile.source_url, model_id))
-        .order_by(LibraryFile.id)
+    readable = owned_row_clauses(LibraryFile, actor, Permission.LIBRARY_READ_ALL, Permission.LIBRARY_READ_OWN)
+    rows = (
+        await db.execute(
+            LibraryFile.active()
+            .where(manyfold_provider.source_url_filter(LibraryFile.source_url, model_id), *readable)
+            .order_by(LibraryFile.id)
+        )
+        if readable is not None
+        else None
     )
-    for row in rows.scalars().all():
+    for row in rows.scalars().all() if rows is not None else []:
         file_id = (row.source_url or "").rsplit("/", 1)[-1]
         imported.setdefault(file_id, ManyfoldLibraryRef(id=row.id, filename=row.filename, folder_id=row.folder_id))
 
@@ -319,9 +327,15 @@ async def import_file(
     source_url = manyfold_provider.canonical_url(
         ProviderResourceRef(source_type=manyfold_provider.source_type, external_id=model_id, sub_id=file_id)
     )
+    # A copy the caller cannot open is no copy for them: they get their own.
+    readable = owned_row_clauses(LibraryFile, actor, Permission.LIBRARY_READ_ALL, Permission.LIBRARY_READ_OWN)
     existing = (
-        await db.execute(LibraryFile.active().where(LibraryFile.source_url == source_url).limit(1))
-    ).scalar_one_or_none()
+        (
+            await db.execute(LibraryFile.active().where(LibraryFile.source_url == source_url, *readable).limit(1))
+        ).scalar_one_or_none()
+        if readable is not None
+        else None
+    )
     if existing is not None:
         return ManyfoldImportResponse(
             library_file_id=existing.id, filename=existing.filename, folder_id=existing.folder_id, was_existing=True
@@ -355,6 +369,7 @@ async def import_file(
         source_type=manyfold_provider.source_type,
         source_url=source_url,
         owner_id=actor.id if actor else None,
+        dedupe_within=readable,
     )
     logger.info(
         "[MANYFOLD] Imported %s (model %s, file %s) as library file %s", filename, model_id, file_id, library_file.id

backend/app/api/routes/pipeline_runs.py

@@ -46,6 +46,7 @@ from backend.app.core.config import settings as app_settings
 from backend.app.core.database import async_session, get_db
 from backend.app.core.permissions import Permission
 from backend.app.core.printer_scope import ALL_PRINTERS, PrinterScope, ensure_model_target_allowed
+from backend.app.core.row_visibility import may_reach_row
 from backend.app.core.websocket import ws_manager
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile
@@ -247,7 +248,11 @@ def _roll_up_run_status(
     return persisted
 
 
-async def _materialise_run(db: AsyncSession, run: PipelineRun) -> PipelineRunResponse:
+async def _materialise_run(
+    db: AsyncSession, run: PipelineRun, actor: User | ApiKeyActor | None = None
+) -> PipelineRunResponse:
+    """The run as ``actor`` sees it: its source and queue jobs are named only
+    when ``actor`` may see them where they live. No actor names everything."""
     pipeline_name: str | None = None
     target_kind = None
     target_printer_id = None
@@ -265,16 +270,28 @@ async def _materialise_run(db: AsyncSession, run: PipelineRun) -> PipelineRunRes
             fanout_strategy = pipeline.fanout_strategy  # type: ignore[assignment]
 
     source_filename: str | None = None
+    source_library_file_id = run.source_library_file_id
+    source_archive_id = run.source_archive_id
     if run.source_library_file_id:
         src = (
             await db.execute(select(LibraryFile).where(LibraryFile.id == run.source_library_file_id))
         ).scalar_one_or_none()
-        source_filename = src.filename if src else None
+        if src is not None and may_reach_row(
+            src.created_by_id, actor, Permission.LIBRARY_READ_ALL, Permission.LIBRARY_READ_OWN
+        ):
+            source_filename = src.filename
+        elif src is not None:
+            source_library_file_id = None
     elif run.source_archive_id:
         arc = (
             await db.execute(select(PrintArchive).where(PrintArchive.id == run.source_archive_id))
         ).scalar_one_or_none()
-        source_filename = (arc.print_name or arc.filename) if arc else None
+        if arc is not None and may_reach_row(
+            arc.created_by_id, actor, Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN
+        ):
+            source_filename = arc.print_name or arc.filename
+        elif arc is not None:
+            source_archive_id = None
 
     job_rows = (
         (
@@ -309,7 +326,10 @@ async def _materialise_run(db: AsyncSession, run: PipelineRun) -> PipelineRunRes
                 copy_index=job.copy_index,
                 assigned_printer_id=job.assigned_printer_id,
                 assigned_printer_name=printer_name,
-                queue_entry_id=job.queue_entry_id,
+                queue_entry_id=job.queue_entry_id
+                if queue_entry is None
+                or may_reach_row(queue_entry.created_by_id, actor, Permission.QUEUE_READ_ALL, Permission.QUEUE_READ_OWN)
+                else None,
                 status=live_job_status,  # type: ignore[arg-type]
                 error_message=job.error_message,
                 dispatched_at=job.dispatched_at,
@@ -323,8 +343,8 @@ async def _materialise_run(db: AsyncSession, run: PipelineRun) -> PipelineRunRes
         id=run.id,
         pipeline_id=run.pipeline_id,
         pipeline_name=pipeline_name,
-        source_library_file_id=run.source_library_file_id,
-        source_archive_id=run.source_archive_id,
+        source_library_file_id=source_library_file_id,
+        source_archive_id=source_archive_id,
         source_filename=source_filename,
         parent_run_id=run.parent_run_id,
         copies=run.copies,
@@ -546,6 +566,9 @@ def _make_orchestration_callable(
                     request=slice_request,
                     current_user_id=creator_user_id,
                     job_id=slice_job_id,
+                    # The run's owner, also when a key without the cloud scope
+                    # started it and no one resolves cloud presets.
+                    owner_id=run.created_by,
                 )
             except HTTPException as exc:
                 run.status = "failed"
@@ -787,7 +810,10 @@ async def run_pipeline(
         copies=body.copies,
         status="queued",
         eligibility_overridden=(not report.ok and body.force),
-        created_by=creator.id if creator else None,
+        # The run, and the queue jobs it adds, are the key's owner's when a key
+        # starts it, as a job the webhook queues is (``creator`` is only who
+        # resolves cloud presets).
+        created_by=creator.id if creator else (actor.id if isinstance(actor, ApiKeyActor) else None),
     )
     db.add(run)
     await db.flush()
@@ -821,7 +847,7 @@ async def run_pipeline(
         kind="library_file" if src_kind == "library_file" else "archive",
         source_id=src_id,
         source_name=src_filename,
-        owner_id=creator.id if creator else None,
+        owner_id=run.created_by,
         run=orchestrate,
     )
 
@@ -829,7 +855,7 @@ async def run_pipeline(
     await db.commit()
     await db.refresh(run)
 
-    return await _materialise_run(db, run)
+    return await _materialise_run(db, run, actor)
 
 
 # ---------------------------------------------------------------------------
@@ -871,6 +897,7 @@ async def list_runs_for_pipeline(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_READ),
     db: AsyncSession = Depends(get_db, scope="function"),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     limit = max(1, min(limit, 100))
     conditions = [PipelineRun.pipeline_id == pipeline_id]
@@ -883,7 +910,7 @@ async def list_runs_for_pipeline(
     )
     total = (await db.execute(select(func.count()).select_from(PipelineRun).where(*conditions))).scalar() or 0
     return PipelineRunListResponse(
-        runs=[await _materialise_run(db, r) for r in rows],
+        runs=[await _materialise_run(db, r, actor) for r in rows],
         total=total,
     )
 
@@ -899,6 +926,7 @@ async def list_all_runs(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_READ),
     db: AsyncSession = Depends(get_db, scope="function"),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Dashboard list. Newest first; filters on pipeline_id + status +
     target_printer_id + target_model_class. The ``status`` filter matches
@@ -934,7 +962,7 @@ async def list_all_runs(
     total = (await db.execute(count_stmt)).scalar() or 0
 
     return PipelineRunListResponse(
-        runs=[await _materialise_run(db, r) for r in rows],
+        runs=[await _materialise_run(db, r, actor) for r in rows],
         total=total,
     )
 
@@ -947,8 +975,9 @@ async def clear_terminal_runs(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_WRITE),
     db: AsyncSession = Depends(get_db, scope="function"),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
-    """Delete every terminal pipeline run (completed / failed / cancelled /
+    """Delete every terminal pipeline run the caller could cancel (completed / failed / cancelled /
     partial_failure). In-flight runs (queued / slicing / dispatching /
     in_progress) are preserved — clearing those mid-flight would lose the
     operator's intent. Cascades to PipelineJob via the ondelete='CASCADE'
@@ -958,9 +987,13 @@ async def clear_terminal_runs(
     # under the same session/transaction as the delete so the numbers can't
     # drift if another caller races in.
     # Runs on printers the caller can't see are left alone (#1727)
-    conditions = [PipelineRun.status.in_(_TERMINAL_RUN_STATUSES)]
+    conditions: list = [PipelineRun.status.in_(_TERMINAL_RUN_STATUSES)]
     if (clause := _run_scope_clause(printer_scope)) is not None:
         conditions.append(clause)
+    # Other users' runs only for someone who may change everyone's queue jobs
+    user = actor.owner if isinstance(actor, ApiKeyActor) else actor
+    if actor is not None and user is not None and not user.has_permission(Permission.QUEUE_UPDATE_ALL.value):
+        conditions.append(PipelineRun.created_by == user.id)
     count_stmt = select(func.count()).select_from(PipelineRun).where(*conditions)
     n = (await db.execute(count_stmt)).scalar() or 0
     if n > 0:
@@ -975,9 +1008,10 @@ async def get_run(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_READ),
     db: AsyncSession = Depends(get_db, scope="function"),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     run = await _load_run_in_scope(db, run_id, printer_scope)
-    return await _materialise_run(db, run)
+    return await _materialise_run(db, run, actor)
 
 
 @pipeline_run_router.post("/{run_id}/cancel", response_model=PipelineRunResponse)
@@ -986,13 +1020,19 @@ async def cancel_run(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PIPELINES_RUN),
     db: AsyncSession = Depends(get_db, scope="function"),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Cancel a queued / in-flight run. Cascades to all non-terminal queue
-    entries; in-flight prints continue on the printer (operator must Stop)."""
+    entries; in-flight prints continue on the printer (operator must Stop).
+
+    The run's queue entries are its starter's: cancelling it takes the right to
+    change their queue jobs, as cancelling one of them on the queue does."""
     run = await _load_run_in_scope(db, run_id, printer_scope)
+    if not may_reach_row(run.created_by, actor, Permission.QUEUE_UPDATE_ALL, Permission.QUEUE_UPDATE_OWN):
+        raise HTTPException(403, "Only the user who started this run can cancel it")
 
     if run.status in ("completed", "failed", "cancelled", "partial_failure"):
-        return await _materialise_run(db, run)
+        return await _materialise_run(db, run, actor)
 
     run.status = "cancelled"
     run.completed_at = datetime.now(timezone.utc)
@@ -1037,7 +1077,7 @@ async def cancel_run(
             scheduler.notify_dispatch_cancelled(entry_id)
     await db.refresh(run)
     await _publish_run_event(db, run)
-    return await _materialise_run(db, run)
+    return await _materialise_run(db, run, actor)
 
 
 @pipeline_run_router.post("/{run_id}/retry-failed", response_model=PipelineRunResponse, status_code=202)
@@ -1107,5 +1147,5 @@ async def retry_failed(
         new_row.parent_run_id = parent.id
         await db.commit()
         await db.refresh(new_row)
-        return await _materialise_run(db, new_row)
+        return await _materialise_run(db, new_row, actor)
     return new_run_response

backend/app/api/routes/projects.py

@@ -33,6 +33,7 @@ from backend.app.core.config import settings
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
 from backend.app.core.printer_scope import PrinterScope
+from backend.app.core.row_visibility import visible_row_clauses
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile, LibraryFolder
 from backend.app.models.print_log import PrintLogEntry
@@ -432,12 +433,14 @@ async def list_projects(
     db: AsyncSession = Depends(get_db, scope="function"),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PROJECTS_READ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """List all projects with basic stats."""
-    # Archive previews show only archives of the caller's printers, as the
-    # project's archive list does.
-    scope_clause = printer_scope.where(PrintArchive.printer_id)
-    preview_scope = [scope_clause] if scope_clause is not None else []
+    # Archive previews show only the archives the project's archive list shows
+    # the caller; the counts and totals cover every print.
+    preview_scope = visible_row_clauses(
+        PrintArchive, printer_scope, actor, Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN
+    )
     query = select(Project)
     if status:
         query = query.where(Project.status == status)
@@ -506,13 +509,15 @@ async def list_projects(
             progress_percent = round((archive_count / project.target_count) * 100, 1)
 
         # Get archive previews (up to 6 most recent)
-        archives_result = await db.execute(
-            select(PrintArchive)
-            .where(PrintArchive.project_id == project.id, _LIVE_ARCHIVE, *preview_scope)
-            .order_by(PrintArchive.created_at.desc())
-            .limit(6)
-        )
-        archives = archives_result.scalars().all()
+        archives = []
+        if preview_scope is not None:
+            archives_result = await db.execute(
+                select(PrintArchive)
+                .where(PrintArchive.project_id == project.id, _LIVE_ARCHIVE, *preview_scope)
+                .order_by(PrintArchive.created_at.desc())
+                .limit(6)
+            )
+            archives = archives_result.scalars().all()
         archive_previews = [
             ArchivePreview(
                 id=a.id,
@@ -959,8 +964,9 @@ async def list_project_archives(
     db: AsyncSession = Depends(get_db, scope="function"),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PROJECTS_READ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
-    """List archives in a project."""
+    """List the archives of a project the caller may see in the archives."""
     # Verify project exists
     result = await db.execute(select(Project).where(Project.id == project_id))
     if not result.scalar_one_or_none():
@@ -980,10 +986,12 @@ async def list_project_archives(
         .limit(limit)
         .offset(offset)
     )
-    # Only archives from printers the caller may see (#1727)
-    if (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
-        query = query.where(clause)
-    result = await db.execute(query)
+    visible = visible_row_clauses(
+        PrintArchive, printer_scope, actor, Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN
+    )
+    if visible is None:
+        return []
+    result = await db.execute(query.where(*visible))
     archives = result.scalars().all()
 
     # Import the response converter from archives module
@@ -1002,17 +1010,25 @@ async def list_project_queue(
     db: AsyncSession = Depends(get_db, scope="function"),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PROJECTS_READ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
-    """List queue items in a project."""
+    """List the queue items of a project the caller may see in the queue."""
     # Verify project exists
     result = await db.execute(select(Project).where(Project.id == project_id))
     if not result.scalar_one_or_none():
         raise HTTPException(status_code=404, detail="Project not found")
 
     # Get queue items
-    query = select(PrintQueueItem).where(PrintQueueItem.project_id == project_id).order_by(PrintQueueItem.position)
-    if (clause := printer_scope.where(PrintQueueItem.printer_id)) is not None:
-        query = query.where(clause)
+    visible = visible_row_clauses(
+        PrintQueueItem, printer_scope, actor, Permission.QUEUE_READ_ALL, Permission.QUEUE_READ_OWN
+    )
+    if visible is None:
+        return []
+    query = (
+        select(PrintQueueItem)
+        .where(PrintQueueItem.project_id == project_id, *visible)
+        .order_by(PrintQueueItem.position)
+    )
     result = await db.execute(query)
     items = result.scalars().all()
 
@@ -1104,26 +1120,14 @@ def _editable_rows(
 ):
     """Select the rows of ``ids`` the caller may relink, or None for none of them.
 
-    The same rows the caller could change through the archive or queue edit:
-    on their printers, and their own unless they may change everyone's. Rows
-    outside are skipped like missing ones, so the count names no hidden id. An
-    API key changes what its owner could. Auth off (no actor) changes all.
+    The same rows the caller could change through the archive or queue edit.
+    Rows outside are skipped like missing ones, so the count names no hidden
+    id. An API key changes what its owner could.
     """
-    query = select(model).where(model.id.in_(ids))
-    if (clause := printer_scope.where(model.printer_id)) is not None:
-        query = query.where(clause)
-    if actor is None:
-        return query
-    # A key acts with its owner's rights to these rows, not with its own scope
-    # flags (a project key needs no archive or queue flag, #1893), so it never
-    # reaches a row its owner could not change. A legacy key without an owner
-    # keeps every row of its printers.
-    user = actor.owner if isinstance(actor, ApiKeyActor) else actor
-    if user is None or user.has_permission(all_permission.value):
-        return query
-    if user.has_permission(own_permission.value):
-        return query.where(model.created_by_id == user.id)
-    return None
+    clauses = visible_row_clauses(model, printer_scope, actor, all_permission, own_permission)
+    if clauses is None:
+        return None
+    return select(model).where(model.id.in_(ids), *clauses)
 
 
 @router.post("/{project_id}/add-archives")
@@ -1366,24 +1370,29 @@ async def delete_project_cover_image(
 # ============ Phase 7: BOM Endpoints ============
 
 
-async def _visible_archive_names(db: AsyncSession, archive_ids, printer_scope: PrinterScope) -> dict[int, str | None]:
+async def _visible_archive_names(
+    db: AsyncSession, archive_ids, printer_scope: PrinterScope, actor: User | ApiKeyActor | None
+) -> dict[int, str | None]:
     """Print names of the archives among ``archive_ids`` the caller can see.
 
-    As the project's archive list: live archives on the caller's printers
-    (#1727). An archive missing from the result reads as missing.
+    As the project's archive list: live archives the caller sees there. An
+    archive missing from the result reads as missing.
     """
     ids = {i for i in archive_ids if i}
-    if not ids:
+    visible = visible_row_clauses(
+        PrintArchive, printer_scope, actor, Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN
+    )
+    if not ids or visible is None:
         return {}
-    query = select(PrintArchive.id, PrintArchive.print_name).where(PrintArchive.id.in_(ids), _LIVE_ARCHIVE)
-    if (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
-        query = query.where(clause)
+    query = select(PrintArchive.id, PrintArchive.print_name).where(PrintArchive.id.in_(ids), _LIVE_ARCHIVE, *visible)
     return dict((await db.execute(query)).all())
 
 
-async def _ensure_linkable_archive(db: AsyncSession, archive_id: int, printer_scope: PrinterScope) -> None:
+async def _ensure_linkable_archive(
+    db: AsyncSession, archive_id: int, printer_scope: PrinterScope, actor: User | ApiKeyActor | None
+) -> None:
     """Refuse a link to an archive the caller cannot see; hidden reads as missing."""
-    if archive_id not in await _visible_archive_names(db, [archive_id], printer_scope):
+    if archive_id not in await _visible_archive_names(db, [archive_id], printer_scope, actor):
         raise HTTPException(status_code=400, detail="Archive not found")
 
 
@@ -1414,6 +1423,7 @@ async def list_bom_items(
     db: AsyncSession = Depends(get_db, scope="function"),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PROJECTS_READ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """List all BOM items for a project."""
     # Verify project exists
@@ -1429,7 +1439,7 @@ async def list_bom_items(
     )
     items = result.scalars().all()
 
-    names = await _visible_archive_names(db, [item.archive_id for item in items], printer_scope)
+    names = await _visible_archive_names(db, [item.archive_id for item in items], printer_scope, actor)
     return [_bom_response(item, names) for item in items]
 
 
@@ -1440,6 +1450,7 @@ async def create_bom_item(
     db: AsyncSession = Depends(get_db, scope="function"),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PROJECTS_UPDATE),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Add a BOM item to a project."""
     # Verify project exists
@@ -1450,7 +1461,7 @@ async def create_bom_item(
     # 0 means no archive, as on update.
     archive_id = data.archive_id or None
     if archive_id is not None:
-        await _ensure_linkable_archive(db, archive_id, printer_scope)
+        await _ensure_linkable_archive(db, archive_id, printer_scope, actor)
 
     # Get max sort order
     max_order_result = await db.execute(
@@ -1473,7 +1484,7 @@ async def create_bom_item(
     await db.flush()
     await db.refresh(item)
 
-    return _bom_response(item, await _visible_archive_names(db, [item.archive_id], printer_scope))
+    return _bom_response(item, await _visible_archive_names(db, [item.archive_id], printer_scope, actor))
 
 
 @router.patch("/{project_id}/bom/{item_id}", response_model=BOMItemResponse)
@@ -1484,6 +1495,7 @@ async def update_bom_item(
     db: AsyncSession = Depends(get_db, scope="function"),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PROJECTS_UPDATE),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Update a BOM item."""
     result = await db.execute(
@@ -1511,7 +1523,7 @@ async def update_bom_item(
         # Also when unchanged: a hidden current link must answer as a missing one,
         # or resending its id would confirm it.
         if data.archive_id != 0:
-            await _ensure_linkable_archive(db, data.archive_id, printer_scope)
+            await _ensure_linkable_archive(db, data.archive_id, printer_scope, actor)
         item.archive_id = data.archive_id if data.archive_id != 0 else None
     if data.stl_filename is not None:
         item.stl_filename = data.stl_filename if data.stl_filename else None
@@ -1521,7 +1533,7 @@ async def update_bom_item(
     await db.flush()
     await db.refresh(item)
 
-    return _bom_response(item, await _visible_archive_names(db, [item.archive_id], printer_scope))
+    return _bom_response(item, await _visible_archive_names(db, [item.archive_id], printer_scope, actor))
 
 
 @router.delete("/{project_id}/bom/{item_id}")
@@ -1660,8 +1672,10 @@ async def get_project_timeline(
     db: AsyncSession = Depends(get_db, scope="function"),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.PROJECTS_READ),
     printer_scope: PrinterScope = RequestPrinterScope,
+    actor: User | ApiKeyActor | None = RequestActor,
 ):
-    """Get timeline of events for a project."""
+    """Get timeline of events for a project, from the archives and queue items
+    the caller may see there."""
     # Verify project exists
     result = await db.execute(select(Project).where(Project.id == project_id))
     project = result.scalar_one_or_none()
@@ -1680,18 +1694,19 @@ async def get_project_timeline(
         )
     )
 
-    # Get archives and add events
-    archive_query = (
-        select(PrintArchive)
-        .where(PrintArchive.project_id == project_id, _LIVE_ARCHIVE)
-        .order_by(PrintArchive.created_at.desc())
-        .limit(limit)
+    # Get archives and add events, the ones the project's archive list shows
+    visible_archives = visible_row_clauses(
+        PrintArchive, printer_scope, actor, Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN
     )
-    # Only events from printers the caller may see, as the archive and queue
-    # lists of the project (#1727)
-    if (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
-        archive_query = archive_query.where(clause)
-    archives = (await db.execute(archive_query)).scalars().all()
+    archives = []
+    if visible_archives is not None:
+        archive_query = (
+            select(PrintArchive)
+            .where(PrintArchive.project_id == project_id, _LIVE_ARCHIVE, *visible_archives)
+            .order_by(PrintArchive.created_at.desc())
+            .limit(limit)
+        )
+        archives = (await db.execute(archive_query)).scalars().all()
 
     for archive in archives:
         if archive.status == "completed":
@@ -1719,21 +1734,24 @@ async def get_project_timeline(
                 )
             )
 
-    # Get queue items
-    queue_query = (
-        select(PrintQueueItem)
-        .where(PrintQueueItem.project_id == project_id)
-        .options(
-            selectinload(PrintQueueItem.archive),
-            selectinload(PrintQueueItem.library_file),
-            selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file),
-        )
-        .order_by(PrintQueueItem.created_at.desc())
-        .limit(limit)
+    # Get queue items, the ones the project's queue list shows
+    visible_items = visible_row_clauses(
+        PrintQueueItem, printer_scope, actor, Permission.QUEUE_READ_ALL, Permission.QUEUE_READ_OWN
     )
-    if (clause := printer_scope.where(PrintQueueItem.printer_id)) is not None:
-        queue_query = queue_query.where(clause)
-    queue_items = (await db.execute(queue_query)).scalars().all()
+    queue_items = []
+    if visible_items is not None:
+        queue_query = (
+            select(PrintQueueItem)
+            .where(PrintQueueItem.project_id == project_id, *visible_items)
+            .options(
+                selectinload(PrintQueueItem.archive),
+                selectinload(PrintQueueItem.library_file),
+                selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file),
+            )
+            .order_by(PrintQueueItem.created_at.desc())
+            .limit(limit)
+        )
+        queue_items = (await db.execute(queue_query)).scalars().all()
 
     for item in queue_items:
         if item.status == "printing":

backend/app/api/routes/webhook.py

@@ -2,7 +2,7 @@ import logging
 
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
-from sqlalchemy import select
+from sqlalchemy import or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.api.routes.print_queue import _assert_can_queue_archive, saved_slicer_ams_mapping
@@ -17,6 +17,8 @@ from backend.app.core.auth import (
     resolve_apikey_owner,
 )
 from backend.app.core.database import get_db
+from backend.app.core.permissions import Permission
+from backend.app.core.row_visibility import may_reach_row
 from backend.app.models.api_key import APIKey
 from backend.app.models.archive import PrintArchive
 from backend.app.models.print_queue import PrintQueueItem
@@ -104,6 +106,12 @@ def _job_id(subtask_id) -> str | None:
 # Webhook endpoints
 
 
+async def _key_actor(db: AsyncSession, api_key: APIKey) -> ApiKeyActor | None:
+    """The key acting for its owner; None with auth off, where there is no owner
+    to act for (as on POST /queue/) and the key's printers still apply."""
+    return ApiKeyActor(api_key, await resolve_apikey_owner(db, api_key)) if await is_auth_enabled(db) else None
+
+
 @router.post("/queue/add", response_model=QueueAddResponse)
 async def webhook_add_to_queue(
     data: QueueAddRequest,
@@ -123,8 +131,7 @@ async def webhook_add_to_queue(
     archive = result.scalar_one_or_none()
     if not archive:
         raise HTTPException(status_code=404, detail="Archive not found")
-    # Auth off: no owner to act for, as on POST /queue/; the key's printers still apply.
-    actor = ApiKeyActor(api_key, await resolve_apikey_owner(db, api_key)) if await is_auth_enabled(db) else None
+    actor = await _key_actor(db, api_key)
     _assert_can_queue_archive(archive, actor, await api_key_printer_scope(db, api_key))
 
     # Verify printer exists
@@ -214,17 +221,31 @@ async def webhook_start_print(
     if not printer:
         raise HTTPException(status_code=404, detail="Printer not found")
 
-    # Get next pending queue item
-    result = await db.execute(
-        select(PrintQueueItem)
-        .where(
-            PrintQueueItem.printer_id == printer_id,
-            PrintQueueItem.status == "pending",
+    # The next pending queue item the key's owner could start from the queue:
+    # theirs, or one nobody owns yet (a virtual-printer upload, claimed by
+    # starting it, #1670), unless they may start everyone's.
+    actor = await _key_actor(db, api_key)
+    owner = actor.owner if actor is not None else None
+    startable: list | None = []
+    if owner is not None and not owner.has_permission(Permission.QUEUE_UPDATE_ALL.value):
+        startable = (
+            [or_(PrintQueueItem.created_by_id == owner.id, PrintQueueItem.created_by_id.is_(None))]
+            if owner.has_permission(Permission.QUEUE_UPDATE_OWN.value)
+            else None
         )
-        .order_by(PrintQueueItem.position)
-        .limit(1)
-    )
-    queue_item = result.scalar_one_or_none()
+    queue_item = None
+    if startable is not None:
+        result = await db.execute(
+            select(PrintQueueItem)
+            .where(
+                PrintQueueItem.printer_id == printer_id,
+                PrintQueueItem.status == "pending",
+                *startable,
+            )
+            .order_by(PrintQueueItem.position)
+            .limit(1)
+        )
+        queue_item = result.scalar_one_or_none()
     if not queue_item:
         raise HTTPException(status_code=404, detail="No pending prints in queue")
 
@@ -244,6 +265,8 @@ async def webhook_start_print(
     # already auto-dispatchable this is a no-op; the scheduler will still
     # pick it up on its next tick.
     queue_item.manual_start = False
+    if owner is not None and queue_item.created_by_id is None:
+        queue_item.created_by_id = owner.id
     await db.commit()
     await db.refresh(queue_item)
 
@@ -382,9 +405,11 @@ async def webhook_get_queue_status(
 ):
     """Get queue status for all printers or a specific printer.
 
-    Requires 'can_read_status' permission.
+    Requires 'can_read_status' permission. The counts cover every job; the
+    items listed are those the key's owner sees in the queue.
     """
     await check_webhook_permission(db, api_key, "read_status")
+    actor = await _key_actor(db, api_key)
 
     # Get printers
     if printer_id:
@@ -428,6 +453,7 @@ async def webhook_get_queue_status(
                         "status": item.status,
                     }
                     for item in items
+                    if may_reach_row(item.created_by_id, actor, Permission.QUEUE_READ_ALL, Permission.QUEUE_READ_OWN)
                 ],
             )
         )

backend/app/core/row_visibility.py

@@ -0,0 +1,68 @@
+"""Which owned rows (archives, queue jobs, library files) a caller reaches through something that holds them."""
+
+from backend.app.core.auth import ApiKeyActor
+from backend.app.core.permissions import Permission
+from backend.app.core.printer_scope import PrinterScope
+from backend.app.models.archive import PrintArchive
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.user import User
+
+
+def owned_row_clauses(
+    model, actor: User | ApiKeyActor | None, all_permission: Permission, own_permission: Permission
+) -> list | None:
+    """WHERE clauses for the rows of ``model`` the caller may reach by ownership, or None for none.
+
+    Their own unless they may reach everyone's (a row without an owner then
+    stays out). Auth off (no actor) reaches all.
+    """
+    if actor is None:
+        return []
+    # A key acts with its owner's rights to these rows, not with its own scope
+    # flags (a project key needs no archive or queue flag, #1893), so it never
+    # reaches a row its owner could not. A legacy key without an owner keeps
+    # every row.
+    user = actor.owner if isinstance(actor, ApiKeyActor) else actor
+    if user is None or user.has_permission(all_permission.value):
+        return []
+    if user.has_permission(own_permission.value):
+        return [model.created_by_id == user.id]
+    return None
+
+
+def visible_row_clauses(
+    model: type[PrintArchive] | type[PrintQueueItem],
+    printer_scope: PrinterScope,
+    actor: User | ApiKeyActor | None,
+    all_permission: Permission,
+    own_permission: Permission,
+) -> list | None:
+    """WHERE clauses for the archive or queue rows the caller may reach, or None for none.
+
+    The rows the caller reaches through the archives or the queue themselves:
+    on their printers, and by ``owned_row_clauses``.
+    """
+    owned = owned_row_clauses(model, actor, all_permission, own_permission)
+    if owned is None:
+        return None
+    if (clause := printer_scope.where(model.printer_id)) is not None:
+        return [clause, *owned]
+    return owned
+
+
+def may_reach_row(
+    created_by_id: int | None,
+    actor: User | ApiKeyActor | None,
+    all_permission: Permission,
+    own_permission: Permission,
+) -> bool:
+    """Whether the caller may reach one row owned by ``created_by_id``, printer scope aside.
+
+    The rule of ``visible_row_clauses`` for a row already loaded.
+    """
+    if actor is None:
+        return True
+    user = actor.owner if isinstance(actor, ApiKeyActor) else actor
+    if user is None or user.has_permission(all_permission.value):
+        return True
+    return user.has_permission(own_permission.value) and created_by_id is not None and created_by_id == user.id

backend/app/core/websocket.py

@@ -134,6 +134,29 @@ class ConnectionManager:
                 if conn in self.active_connections:
                     self.active_connections.remove(conn)
 
+    async def broadcast_to_users(self, user_ids: set[int], message: dict[str, Any]):
+        """Send a message to every connection of any of ``user_ids``, as
+        ``broadcast_to_user`` does for one."""
+        if not self.active_connections:
+            return
+
+        data = json.dumps(message)
+        printer_id = _message_printer_id(message)
+        async with self._lock:
+            disconnected = []
+            for connection in self.active_connections:
+                conn_uid = getattr(connection.state, "bambuddy_principal_user_id", None)
+                if conn_uid is None or conn_uid not in user_ids or not _may_receive(connection, printer_id):
+                    continue
+                try:
+                    await connection.send_text(data)
+                except Exception:
+                    disconnected.append(connection)
+
+            for conn in disconnected:
+                if conn in self.active_connections:
+                    self.active_connections.remove(conn)
+
     async def close_user(self, user_id: int):
         """Close a deleted user's sockets, which would otherwise keep receiving
         the messages of whoever is given their id next."""
@@ -262,22 +285,28 @@ class ConnectionManager:
             }
         )
 
-    async def send_print_confirm_request(self, printer_id: int, data: dict):
-        """Ask connected clients for a post-print outcome verdict (#1898)."""
-        await self.broadcast(
-            {
-                "type": "print_confirm_request",
-                "printer_id": printer_id,
-                "data": data,
-            }
-        )
+    async def send_print_confirm_request(self, printer_id: int, data: dict, user_ids: set[int] | None = None):
+        """Ask connected clients for a post-print outcome verdict (#1898).
+
+        ``user_ids`` are the users who may answer it; None asks everyone who
+        can see the printer.
+        """
+        message = {"type": "print_confirm_request", "printer_id": printer_id, "data": data}
+        if user_ids is None:
+            await self.broadcast(message)
+        else:
+            await self.broadcast_to_users(user_ids, message)
 
     async def send_archive_created(self, archive: dict):
-        """Notify clients that a new archive was created."""
+        """Notify clients that a new archive was created.
+
+        Only that one was, on which printer: whose it is and what it holds stays
+        with the archive routes, which every socket's client asks next anyway.
+        """
         await self.broadcast(
             {
                 "type": "archive_created",
-                "data": archive,
+                "data": {"printer_id": archive.get("printer_id")},
             }
         )
 
@@ -286,11 +315,12 @@ class ConnectionManager:
 
         Only sockets that may see the archive's printer receive it: callers send
         just the changed fields, so the printer is looked up here when missing.
+        As for a new archive, the message says no more than that.
         """
-        message: dict[str, Any] = {"type": "archive_updated", "data": archive}
-        if _message_printer_id(message) is None:
-            message["printer_id"] = await _archive_printer_id(archive.get("id"))
-        await self.broadcast(message)
+        printer_id = archive.get("printer_id")
+        if not isinstance(printer_id, int):
+            printer_id = await _archive_printer_id(archive.get("id"))
+        await self.broadcast({"type": "archive_updated", "printer_id": printer_id, "data": {"printer_id": printer_id}})
 
     async def send_queue_item_uploading(
         self,

backend/app/main.py

@@ -4311,6 +4311,22 @@ async def _dispatch_outcome_confirmation_safely(
             pass
 
 
+async def _outcome_answerers(db, owner_id: int | None) -> set[int] | None:
+    """Who may answer a print's outcome prompt in the app: its owner and anyone
+    who may change everyone's archives. None (everyone on the printer) with auth
+    off, or for a print nobody owns."""
+    from sqlalchemy.orm import selectinload
+
+    from backend.app.core.auth import is_auth_enabled
+    from backend.app.core.permissions import Permission
+    from backend.app.models.user import User
+
+    if owner_id is None or not await is_auth_enabled(db):
+        return None
+    users = (await db.execute(select(User).options(selectinload(User.groups)).where(User.is_active))).scalars().all()
+    return {owner_id} | {u.id for u in users if u.has_permission(Permission.ARCHIVES_UPDATE_ALL.value)}
+
+
 async def dispatch_outcome_confirmation(
     db,
     printer_id: int,
@@ -4377,6 +4393,7 @@ async def dispatch_outcome_confirmation(
             "archive_id": archive_id,
             "print_name": confirm_archive.print_name or confirm_archive.filename,
         },
+        user_ids=await _outcome_answerers(db, confirm_archive.created_by_id),
     )
     await notification_service.on_print_confirm_request(
         printer_id,

backend/tests/integration/test_api_key_queue_acts_as_owner.py

@@ -10,6 +10,7 @@ nothing and may use no cost center.
 from __future__ import annotations
 
 from pathlib import Path
+from unittest.mock import AsyncMock, patch
 from uuid import uuid4
 
 import pytest
@@ -529,3 +530,51 @@ class TestArchivesAndPipelines:
         # Refused by the ownership check, not later for the missing file
         assert response.status_code == 404
         assert response.json()["detail"] == "File not found"
+
+
+class TestSlicedThroughAKeyIsTheOwners:
+    """A key without the cloud scope resolves no cloud presets, but what it
+    slices is still its owner's, as a pipeline run it starts is."""
+
+    async def _slice(self, async_client, url, key):
+        from dataclasses import dataclass
+
+        @dataclass
+        class _Job:
+            id: int = 1
+            status: str = "pending"
+
+        with patch(
+            "backend.app.services.slice_dispatch.slice_dispatch.enqueue", new=AsyncMock(return_value=_Job())
+        ) as enqueue:
+            response = await async_client.post(url, json=_SLICE_BODY, headers=key)
+        assert response.status_code == 202, response.text
+        return enqueue.await_args.kwargs["owner_id"]
+
+    async def test_a_library_file(self, async_client, db_session, library_world, tmp_path, monkeypatch):
+        monkeypatch.setattr(app_settings, "base_dir", tmp_path)
+        (tmp_path / "mine.3mf").write_bytes(b"PK")
+        mine = LibraryFile(
+            filename="mine.3mf",
+            file_path="mine.3mf",
+            file_type="3mf",
+            file_size=2,
+            created_by_id=library_world["owner"].id,
+        )
+        db_session.add(mine)
+        await db_session.commit()
+
+        owner = await self._slice(async_client, f"/api/v1/library/files/{mine.id}/slice", library_world["key"])
+
+        assert owner == library_world["owner"].id
+
+    async def test_an_archive(self, async_client, db_session, library_world, tmp_path, monkeypatch):
+        monkeypatch.setattr(app_settings, "base_dir", tmp_path)
+        (tmp_path / "mine_archive.3mf").write_bytes(b"PK")
+        archive = await _archive(db_session, library_world["owner"].id, 77)
+        archive.file_path = "mine_archive.3mf"
+        await db_session.commit()
+
+        owner = await self._slice(async_client, f"/api/v1/archives/{archive.id}/slice", library_world["key"])
+
+        assert owner == library_world["owner"].id

backend/tests/integration/test_bulk_reads_api.py

@@ -429,7 +429,9 @@ class TestArchiveFolders:
         )
         await db_session.commit()
         admin = await _admin_token(async_client)
-        jwt, _ = await _user(async_client, admin, "member", ["library:read_all"], printer_ids=[mine.id])
+        jwt, _ = await _user(
+            async_client, admin, "member", ["library:read_all", "archives:read_all"], printer_ids=[mine.id]
+        )
 
         bulk = await _get(
             async_client,

backend/tests/integration/test_external_print_confirmation_1898.py

@@ -582,6 +582,8 @@ class TestTheCompletionEmitsThePrompt:
         assert sent is True
         ws.send_print_confirm_request.assert_awaited_once()
         assert ws.send_print_confirm_request.await_args.args[1]["archive_id"] == archive_id
+        # Nobody owns an external print: everyone on the printer is asked
+        assert ws.send_print_confirm_request.await_args.kwargs["user_ids"] is None
 
         notif.on_print_confirm_request.assert_awaited_once()
         kwargs = notif.on_print_confirm_request.await_args.kwargs
@@ -659,3 +661,32 @@ class TestTheCompletionEmitsThePrompt:
 
         assert sent is False
         notif.on_print_confirm_request.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.parametrize("auth_on", [True, False])
+async def test_an_owned_print_asks_its_owner_and_who_may_answer_for_everyone(
+    test_engine, db_session, printer_factory, archive_factory, auth_on
+):
+    from backend.app.models.group import Group
+    from backend.app.models.user import User
+
+    printer = await printer_factory()
+    if auth_on:
+        await _set_setting(db_session, "auth_enabled", "true")
+    staff = Group(name="staff", permissions=["archives:update_all"])
+    others = Group(name="others", permissions=["archives:update_own"])
+    owner = User(username="prompted", password_hash="x", role="user", groups=[others])
+    manager = User(username="manager", password_hash="x", role="user", groups=[staff])
+    bystander = User(username="bystander", password_hash="x", role="user", groups=[others])
+    gone = User(username="gone", password_hash="x", role="user", groups=[staff], is_active=False)
+    db_session.add_all([staff, others, owner, manager, bystander, gone])
+    await db_session.commit()
+    archive = await archive_factory(printer.id, status="completed", created_by_id=owner.id, confirm_requested=True)
+
+    sent, ws, _ = await _dispatch(db_session, printer.id, archive.id)
+
+    assert sent is True
+    asked = ws.send_print_confirm_request.await_args.kwargs["user_ids"]
+    assert asked == ({owner.id, manager.id} if auth_on else None)

backend/tests/integration/test_library_folder_archive_scope.py

@@ -9,7 +9,7 @@ from backend.tests.integration.test_printer_scope_1727 import _admin_token, _aut
 
 pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
 
-PERMS = ["printers:read", "library:read_all", "library:upload", "library:update_all"]
+PERMS = ["printers:read", "archives:read_all", "library:read_all", "library:upload", "library:update_all"]
 
 
 async def _setup(client: AsyncClient, printer_factory, archive_factory):

backend/tests/integration/test_makerworld_permission_gate.py

@@ -178,3 +178,43 @@ class TestPermissionStillBites:
             )
         assert resp.status_code == 200, resp.text
         assert resp.json()["model_id"] == 1400373
+
+
+class TestAnotherUsersImport:
+    """A user who sees only their own files neither sees a colleague's import
+    nor gets it back for their own: they import a copy of their own."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_is_not_theirs(self, async_client: AsyncClient):
+        admin = await _setup_auth_with_admin(async_client)
+        perms = ["makerworld:view", "makerworld:import", "library:read_own", "library:upload"]
+        first = {"Authorization": f"Bearer {await _make_user(async_client, admin, username='mwa', permissions=perms)}"}
+        second = {"Authorization": f"Bearer {await _make_user(async_client, admin, username='mwb', permissions=perms)}"}
+
+        def resolver():
+            return _fake_service(
+                resolve=ProviderResolvedModel(
+                    ref=ProviderResourceRef(source_type="makerworld", external_id="1400373"),
+                    design={"id": 1400373},
+                    instances=[],
+                )
+            )
+
+        body = {"model_id": 1400373, "profile_id": 298919107}
+        with patch(
+            "backend.app.api.routes.makerworld._build_service", AsyncMock(side_effect=lambda *a: _import_service())
+        ):
+            theirs = (await async_client.post("/api/v1/makerworld/import", json=body, headers=first)).json()
+        with patch("backend.app.api.routes.makerworld._build_service", AsyncMock(side_effect=lambda *a: resolver())):
+            seen = await async_client.post(
+                "/api/v1/makerworld/resolve", json={"url": "https://makerworld.com/en/models/1400373"}, headers=second
+            )
+        with patch(
+            "backend.app.api.routes.makerworld._build_service", AsyncMock(side_effect=lambda *a: _import_service())
+        ):
+            mine = (await async_client.post("/api/v1/makerworld/import", json=body, headers=second)).json()
+
+        assert seen.json()["already_imported_library_ids"] == []
+        assert mine["was_existing"] is False
+        assert mine["library_file_id"] != theirs["library_file_id"]

backend/tests/integration/test_manyfold_api.py

@@ -335,6 +335,27 @@ class TestPermissions:
         assert (await async_client.get(f"{API}/models", headers=mw)).status_code == 403
         assert (await async_client.get(f"{API}/status", headers=mw)).status_code == 403
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_another_users_import_is_not_theirs(self, async_client, manyfold):
+        """A user who sees only their own files neither sees a colleague's import
+        nor gets it back for their own: they import a copy of their own."""
+        admin = await _admin(async_client)
+        await _connect(async_client, manyfold, admin)
+        perms = ["manyfold:view", "manyfold:import", "library:read_own", "library:upload"]
+        first = await _user(async_client, admin, "mffirst", perms)
+        second = await _user(async_client, admin, "mfsecond", perms)
+        body = {"model_id": "cube01", "file_id": "f1"}
+
+        theirs = (await async_client.post(f"{API}/import", headers=first, json=body)).json()
+        seen = (await async_client.get(f"{API}/models/cube01", headers=second)).json()
+        mine = (await async_client.post(f"{API}/import", headers=second, json=body)).json()
+        seen_after = (await async_client.get(f"{API}/models/cube01", headers=second)).json()
+
+        assert {f["id"]: f["library_file"] for f in seen["files"]}["f1"] is None
+        assert mine["was_existing"] is False and mine["library_file_id"] != theirs["library_file_id"]
+        assert {f["id"]: f["library_file"] for f in seen_after["files"]}["f1"]["id"] == mine["library_file_id"]
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_anonymous_preview_is_refused(self, async_client, manyfold):

backend/tests/integration/test_owned_rows_elsewhere.py

@@ -0,0 +1,328 @@
+"""Pipeline runs, the webhook and live archive events keep to what each user may see.
+
+A pipeline run's queue jobs and source belong to the user who started it. The
+run shows another user neither its source nor its queue jobs, and only someone
+who may change those jobs cancels or clears it. A webhook key acts for its
+owner: it starts only a job its owner could start, and lists only the jobs its
+owner sees. A live archive event says only that an archive on a printer
+changed.
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from backend.app.core import websocket
+from backend.app.core.printer_scope import ALL_PRINTERS
+from backend.app.models.pipeline_run import PipelineJob, PipelineRun
+from backend.app.models.print_queue import PrintQueueItem
+from backend.tests.integration.test_printer_scope_1727 import _admin_token, _auth
+from backend.tests.integration.test_project_batch_scope import _member
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+OPERATOR = [
+    "printers:read",
+    "printers:control",
+    "api_keys:create",
+    "archives:read_own",
+    "archives:update_own",
+    "queue:read_own",
+    "queue:update_own",
+    "library:read_own",
+    "makerworld:view",
+    "pipelines:read",
+    "pipelines:write",
+    "pipelines:run",
+]
+
+
+async def _job(db_session, printer_id: int, archive_id: int, owner: int | None, position: int = 1) -> int:
+    item = PrintQueueItem(
+        printer_id=printer_id, archive_id=archive_id, created_by_id=owner, status="pending", position=position
+    )
+    db_session.add(item)
+    await db_session.commit()
+    return item.id
+
+
+@pytest.fixture
+async def two(async_client: AsyncClient, printer_factory, archive_factory, db_session):
+    printer = await printer_factory(name="A")
+    admin = await _admin_token(async_client)
+    first, first_id = await _member(async_client, admin, "first", None, OPERATOR)
+    second, second_id = await _member(async_client, admin, "second", None, OPERATOR)
+    return SimpleNamespace(
+        printer=printer, admin=_auth(admin), first=first, first_id=first_id, second=second, second_id=second_id
+    )
+
+
+async def _run(db_session, archive_factory, two, owner_id: int, status: str = "in_progress"):
+    archive = await archive_factory(two.printer.id, created_by_id=owner_id, print_name="Their part")
+    entry = await _job(db_session, two.printer.id, archive.id, owner_id)
+    run = PipelineRun(source_archive_id=archive.id, status=status, copies=1, created_by=owner_id)
+    db_session.add(run)
+    await db_session.commit()
+    db_session.add(
+        PipelineJob(pipeline_run_id=run.id, assigned_printer_id=two.printer.id, queue_entry_id=entry, status="queued")
+    )
+    await db_session.commit()
+    return run.id, archive.id, entry
+
+
+async def _entry_status(db_session, entry: int) -> str:
+    db_session.expire_all()
+    return (await db_session.execute(select(PrintQueueItem.status).where(PrintQueueItem.id == entry))).scalar_one()
+
+
+class TestPipelineRuns:
+    async def test_another_users_run_names_nothing_of_theirs(
+        self, async_client: AsyncClient, two, archive_factory, db_session
+    ):
+        run, archive, entry = await _run(db_session, archive_factory, two, two.second_id)
+
+        theirs = (await async_client.get(f"/api/v1/pipeline-runs/{run}", headers=two.first)).json()
+        listed = (await async_client.get("/api/v1/pipeline-runs", headers=two.first)).json()["runs"]
+        own = (await async_client.get(f"/api/v1/pipeline-runs/{run}", headers=two.second)).json()
+
+        for seen in (theirs, *listed):
+            assert (seen["source_archive_id"], seen["source_filename"]) == (None, None)
+            assert [j["queue_entry_id"] for j in seen["jobs"]] == [None]
+        assert (own["source_archive_id"], own["source_filename"]) == (archive, "Their part")
+        assert [j["queue_entry_id"] for j in own["jobs"]] == [entry]
+
+    async def test_nor_a_library_file_of_theirs(self, async_client: AsyncClient, two, db_session):
+        from backend.app.models.library import LibraryFile
+
+        source = LibraryFile(
+            filename="their.3mf",
+            file_path="their.3mf",
+            file_type="3mf",
+            file_size=0,
+            file_hash="h",
+            source_type="uploaded",
+            created_by_id=two.second_id,
+        )
+        db_session.add(source)
+        await db_session.commit()
+        run = PipelineRun(source_library_file_id=source.id, status="completed", copies=1, created_by=two.second_id)
+        db_session.add(run)
+        await db_session.commit()
+
+        theirs = (await async_client.get(f"/api/v1/pipeline-runs/{run.id}", headers=two.first)).json()
+        own = (await async_client.get(f"/api/v1/pipeline-runs/{run.id}", headers=two.second)).json()
+
+        assert (theirs["source_library_file_id"], theirs["source_filename"]) == (None, None)
+        assert (own["source_library_file_id"], own["source_filename"]) == (source.id, "their.3mf")
+
+    async def test_only_its_starter_cancels_it(self, async_client: AsyncClient, two, archive_factory, db_session):
+        run, _, entry = await _run(db_session, archive_factory, two, two.second_id)
+
+        refused = await async_client.post(f"/api/v1/pipeline-runs/{run}/cancel", headers=two.first)
+        assert refused.status_code == 403
+        assert await _entry_status(db_session, entry) == "pending"
+
+        cancelled = await async_client.post(f"/api/v1/pipeline-runs/{run}/cancel", headers=two.second)
+        assert cancelled.status_code == 200, cancelled.text
+        assert await _entry_status(db_session, entry) == "cancelled"
+
+    async def test_or_someone_who_may_change_every_job(
+        self, async_client: AsyncClient, two, archive_factory, db_session
+    ):
+        run, _, entry = await _run(db_session, archive_factory, two, two.second_id)
+
+        cancelled = await async_client.post(f"/api/v1/pipeline-runs/{run}/cancel", headers=two.admin)
+
+        assert cancelled.status_code == 200, cancelled.text
+        assert await _entry_status(db_session, entry) == "cancelled"
+
+    async def test_clearing_finished_runs_clears_only_ones_own(
+        self, async_client: AsyncClient, two, archive_factory, db_session
+    ):
+        mine, *_ = await _run(db_session, archive_factory, two, two.first_id, status="completed")
+        theirs, *_ = await _run(db_session, archive_factory, two, two.second_id, status="completed")
+
+        cleared = await async_client.post("/api/v1/pipeline-runs/clear", headers=two.first)
+
+        assert cleared.json() == {"deleted": 1}
+        db_session.expire_all()
+        assert list((await db_session.execute(select(PipelineRun.id))).scalars()) == [theirs]
+        assert mine != theirs
+
+
+async def _key(client: AsyncClient, headers: dict, printer_id: int) -> dict:
+    created = await client.post(
+        "/api/v1/api-keys/",
+        headers=headers,
+        json={"name": "k", "can_control_printer": True, "can_read_status": True, "printer_ids": [printer_id]},
+    )
+    assert created.status_code in (200, 201), created.text
+    return {"X-API-Key": created.json()["key"]}
+
+
+class TestWebhook:
+    async def test_start_takes_the_next_job_its_owner_could_start(
+        self, async_client: AsyncClient, two, archive_factory, db_session
+    ):
+        archive = await archive_factory(two.printer.id)
+        await _job(db_session, two.printer.id, archive.id, two.second_id, position=1)
+        mine = await _job(db_session, two.printer.id, archive.id, two.first_id, position=2)
+        key = await _key(async_client, two.first, two.printer.id)
+
+        started = await async_client.post(f"/api/v1/webhook/printer/{two.printer.id}/start", headers=key)
+
+        assert started.status_code == 200, started.text
+        assert started.json()["queue_item_id"] == mine
+
+    async def test_a_job_nobody_owns_is_started_and_becomes_the_owners(
+        self, async_client: AsyncClient, two, archive_factory, db_session
+    ):
+        archive = await archive_factory(two.printer.id)
+        unowned = await _job(db_session, two.printer.id, archive.id, None, position=1)
+        key = await _key(async_client, two.first, two.printer.id)
+
+        started = await async_client.post(f"/api/v1/webhook/printer/{two.printer.id}/start", headers=key)
+
+        assert started.json()["queue_item_id"] == unowned
+        db_session.expire_all()
+        owner = await db_session.scalar(select(PrintQueueItem.created_by_id).where(PrintQueueItem.id == unowned))
+        assert owner == two.first_id
+
+    async def test_and_none_when_only_others_wait(self, async_client: AsyncClient, two, archive_factory, db_session):
+        archive = await archive_factory(two.printer.id)
+        await _job(db_session, two.printer.id, archive.id, two.second_id)
+        key = await _key(async_client, two.first, two.printer.id)
+
+        started = await async_client.post(f"/api/v1/webhook/printer/{two.printer.id}/start", headers=key)
+
+        assert started.status_code == 404
+
+    async def test_queue_status_lists_only_the_owners_jobs(
+        self, async_client: AsyncClient, two, archive_factory, db_session
+    ):
+        archive = await archive_factory(two.printer.id)
+        await _job(db_session, two.printer.id, archive.id, two.second_id, position=1)
+        mine = await _job(db_session, two.printer.id, archive.id, two.first_id, position=2)
+        key = await _key(async_client, two.first, two.printer.id)
+
+        (status,) = (await async_client.get("/api/v1/webhook/queue", headers=key)).json()
+
+        assert status["pending"] == 2
+        assert [item["id"] for item in status["items"]] == [mine]
+
+
+class TestLiveArchiveEvents:
+    @pytest.mark.parametrize("send", ["send_archive_created", "send_archive_updated"])
+    async def test_an_event_names_only_the_printer(self, send):
+        manager = websocket.ConnectionManager()
+        socket = SimpleNamespace(state=SimpleNamespace(bambuddy_printer_scope=ALL_PRINTERS), send_text=AsyncMock())
+        manager.active_connections = [socket]
+
+        await getattr(manager, send)(
+            {"id": 9, "printer_id": 1, "filename": "part.3mf", "print_name": "Part", "status": "printing", "cost": 2}
+        )
+
+        message = json.loads(socket.send_text.await_args.args[0])
+        assert message["data"] == {"printer_id": 1}
+
+
+def _user_socket(user_id: int | None):
+    return SimpleNamespace(
+        state=SimpleNamespace(bambuddy_printer_scope=ALL_PRINTERS, bambuddy_principal_user_id=user_id),
+        send_text=AsyncMock(),
+    )
+
+
+class TestOutcomePrompt:
+    async def test_who_may_answer_is_asked(self):
+        manager = websocket.ConnectionManager()
+        owner, other = _user_socket(7), _user_socket(8)
+        manager.active_connections = [owner, other]
+
+        await manager.send_print_confirm_request(1, {"archive_id": 9, "print_name": "Part"}, user_ids={7})
+
+        owner.send_text.assert_awaited_once()
+        other.send_text.assert_not_awaited()
+
+    async def test_an_archive_nobody_owns_asks_everyone(self):
+        manager = websocket.ConnectionManager()
+        sockets = [_user_socket(7), _user_socket(8)]
+        manager.active_connections = list(sockets)
+
+        await manager.send_print_confirm_request(1, {"archive_id": 9, "print_name": "Part"})
+
+        for socket in sockets:
+            socket.send_text.assert_awaited_once()
+
+
+class TestMakerWorldRecentImports:
+    async def test_only_ones_own_imports_are_listed(self, async_client: AsyncClient, two, db_session):
+        from backend.app.models.library import LibraryFile
+
+        rows = {}
+        for who, owner in (("first", two.first_id), ("second", two.second_id)):
+            row = LibraryFile(
+                filename=f"{who}.3mf",
+                file_path=f"{who}.3mf",
+                file_type="3mf",
+                file_size=0,
+                file_hash=who,
+                source_type="makerworld",
+                source_url=f"https://makerworld.com/models/{who}",
+                created_by_id=owner,
+            )
+            db_session.add(row)
+            await db_session.commit()
+            rows[who] = row.id
+
+        listed = await async_client.get("/api/v1/makerworld/recent-imports", headers=two.first)
+
+        assert listed.status_code == 200, listed.text
+        assert [r["library_file_id"] for r in listed.json()] == [rows["first"]]
+
+
+class TestASecondCopyOnAMount:
+    async def test_is_stored_beside_a_colleagues_under_a_free_name(self, db_session, tmp_path):
+        from backend.app.api.routes.library import save_3mf_bytes_to_library
+        from backend.app.models.library import LibraryFile, LibraryFolder
+
+        mount = tmp_path / "nas"
+        mount.mkdir()
+        (mount / "benchy.stl").write_bytes(b"theirs")
+        folder = LibraryFolder(name="NAS", is_external=True, external_path=str(mount), external_readonly=False)
+        db_session.add(folder)
+        await db_session.commit()
+        db_session.add(
+            LibraryFile(
+                folder_id=folder.id,
+                is_external=True,
+                filename="benchy.stl",
+                file_path=str(mount / "benchy.stl"),
+                file_type="stl",
+                file_size=6,
+                file_hash="t",
+                source_type="manyfold",
+                source_url="manyfold:m1/f1",
+            )
+        )
+        await db_session.commit()
+
+        mine, was_existing = await save_3mf_bytes_to_library(
+            db_session,
+            file_bytes=b"solid x\nendsolid x\n",
+            filename="benchy.stl",
+            folder_id=folder.id,
+            source_type="manyfold",
+            source_url="manyfold:m1/f1",
+            dedupe_within=None,
+        )
+
+        assert was_existing is False
+        assert mine.filename == "benchy (2).stl"
+        assert (mount / "benchy.stl").read_bytes() == b"theirs"

backend/tests/integration/test_pipeline_runs_api.py

@@ -7,7 +7,7 @@ needing a live sidecar in CI.
 
 from __future__ import annotations
 
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from httpx import AsyncClient
@@ -1199,6 +1199,51 @@ class TestRunViaApiKey:
 
         assert resp.status_code == 202, resp.text
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_run_a_key_starts_is_its_owners(
+        self,
+        async_client: AsyncClient,
+        pipeline_factory,
+        printer_factory,
+        library_file_factory,
+        db_session,
+    ):
+        """Without the cloud scope too: whose the run is decides who may cancel it."""
+        from dataclasses import dataclass
+
+        @dataclass
+        class _FakeSliceJob:
+            id: int = 7778
+
+        printer = await printer_factory()
+        pipeline = await pipeline_factory(target_printer_id=printer.id)
+        src = await library_file_factory()
+        admin, key = await self._admin_and_key(
+            async_client, db_session, can_read_status=True, can_queue=True, can_manage_library=True
+        )
+
+        live_status = {"connected": True, "raw_data": {"ams": []}}
+        with (
+            patch(
+                "backend.app.api.routes.pipeline_runs._load_printer_status",
+                new=AsyncMock(return_value=live_status),
+            ),
+            patch(
+                "backend.app.services.slice_dispatch.slice_dispatch.enqueue",
+                new=AsyncMock(return_value=_FakeSliceJob()),
+            ) as enqueue,
+        ):
+            resp = await async_client.post(
+                f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
+                json={"source_library_file_id": src.id, "copies": 1, "force": True},
+                headers={"X-API-Key": key},
+            )
+
+        assert resp.status_code == 202, resp.text
+        assert resp.json()["created_by"] == admin.id
+        assert enqueue.await_args.kwargs["owner_id"] == admin.id
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_a_key_without_manage_library_is_refused(
@@ -1289,7 +1334,8 @@ class TestRunViaApiKey:
         db_session,
     ):
         """The fallback is the cloud scope's own opt-in, not a general identity
-        for API keys: a key without it slices unattributed, exactly as before."""
+        for API keys: a key without it resolves no cloud presets as anyone. The
+        run and its output are still its owner's."""
         from dataclasses import dataclass
 
         @dataclass
@@ -1301,18 +1347,22 @@ class TestRunViaApiKey:
         src = await library_file_factory()
         # Same owner and same three scopes as the test above — only the cloud
         # opt-in differs.
-        _, full_key = await self._admin_and_key(
+        admin, full_key = await self._admin_and_key(
             async_client, db_session, can_read_status=True, can_queue=True, can_manage_library=True
         )
 
+        from backend.app.api.routes import pipeline_runs
+
         enqueue = AsyncMock(return_value=_FakeSliceJob())
         live_status = {"connected": True, "raw_data": {"ams": []}}
+        orchestration = MagicMock(wraps=pipeline_runs._make_orchestration_callable)
         with (
             patch(
                 "backend.app.api.routes.pipeline_runs._load_printer_status",
                 new=AsyncMock(return_value=live_status),
             ),
             patch("backend.app.services.slice_dispatch.slice_dispatch.enqueue", new=enqueue),
+            patch("backend.app.api.routes.pipeline_runs._make_orchestration_callable", new=orchestration),
         ):
             resp = await async_client.post(
                 f"/api/v1/slicer-pipelines/{pipeline['id']}/run",
@@ -1321,7 +1371,8 @@ class TestRunViaApiKey:
             )
 
         assert resp.status_code == 202, resp.text
-        assert enqueue.await_args.kwargs["owner_id"] is None
+        assert orchestration.call_args.kwargs["creator_user_id"] is None
+        assert enqueue.await_args.kwargs["owner_id"] == admin.id
 
 
 class TestJobKeepsItsFinalStatus:

backend/tests/integration/test_project_row_visibility.py

@@ -0,0 +1,227 @@
+"""A project shows each caller the archives and queue jobs they may see.
+
+Projects are shared, but the archives and queue jobs in them belong to the
+users who made them. The project's archive list, queue list and timeline show
+the caller the rows the archives and the queue themselves would show them:
+their own, unless they may see everyone's, on their printers. An API key sees
+what its owner could. With auth off every row shows.
+"""
+
+from __future__ import annotations
+
+import pytest
+from httpx import AsyncClient
+
+from backend.app.models.print_queue import PrintQueueItem
+from backend.tests.integration.test_printer_scope_1727 import _admin_token, _auth
+from backend.tests.integration.test_project_batch_scope import _archive_projects, _member, _project
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+BASE = ["printers:read", "api_keys:create", "projects:read", "projects:update"]
+OWN = [*BASE, "archives:read_own", "archives:update_own", "queue:read_own", "library:read_own", "library:upload"]
+
+
+async def _job(db_session, printer_id: int, archive_id: int, project_id: int, created_by_id: int | None) -> int:
+    item = PrintQueueItem(
+        printer_id=printer_id,
+        archive_id=archive_id,
+        project_id=project_id,
+        created_by_id=created_by_id,
+        status="pending",
+    )
+    db_session.add(item)
+    await db_session.commit()
+    return item.id
+
+
+@pytest.fixture
+async def shared(async_client: AsyncClient, printer_factory, archive_factory, db_session):
+    """A project holding one archive and one job of each of two users, and an ownerless pair."""
+    a = await printer_factory(name="A")
+    admin = await _admin_token(async_client)
+    first, first_id = await _member(async_client, admin, "first", None, OWN)
+    second, second_id = await _member(async_client, admin, "second", None, OWN)
+    project = await _project(async_client, admin)
+    rows = {}
+    for who, user_id in (("first", first_id), ("second", second_id), ("nobody", None)):
+        archive = await archive_factory(
+            a.id, created_by_id=user_id, project_id=project, print_name=f"{who} part", status="completed"
+        )
+        rows[who] = (archive.id, await _job(db_session, a.id, archive.id, project, user_id))
+    return {"admin": _auth(admin), "first": first, "second": second, "project": project, "rows": rows}
+
+
+async def _lists(client: AsyncClient, headers: dict, project: int):
+    archives = await client.get(f"/api/v1/projects/{project}/archives", headers=headers)
+    queue = await client.get(f"/api/v1/projects/{project}/queue", headers=headers)
+    timeline = await client.get(f"/api/v1/projects/{project}/timeline", headers=headers)
+    for response in (archives, queue, timeline):
+        assert response.status_code == 200, response.text
+    events = timeline.json()
+    return (
+        sorted(a["id"] for a in archives.json()),
+        sorted(q["id"] for q in queue.json()),
+        sorted(e["metadata"]["archive_id"] for e in events if "archive_id" in (e.get("metadata") or {})),
+        sorted(e["metadata"]["queue_item_id"] for e in events if "queue_item_id" in (e.get("metadata") or {})),
+    )
+
+
+async def test_a_user_sees_only_their_own_rows(async_client: AsyncClient, shared):
+    archive, job = shared["rows"]["first"]
+
+    assert await _lists(async_client, shared["first"], shared["project"]) == ([archive], [job], [archive], [job])
+
+
+async def test_so_does_the_other_one(async_client: AsyncClient, shared):
+    archive, job = shared["rows"]["second"]
+
+    assert await _lists(async_client, shared["second"], shared["project"]) == ([archive], [job], [archive], [job])
+
+
+async def test_an_admin_sees_every_row(async_client: AsyncClient, shared):
+    archives = sorted(a for a, _ in shared["rows"].values())
+    jobs = sorted(j for _, j in shared["rows"].values())
+
+    assert await _lists(async_client, shared["admin"], shared["project"]) == (archives, jobs, archives, jobs)
+
+
+async def test_without_archive_or_queue_rights_nothing_shows(async_client: AsyncClient, admin_and_editor):
+    headers, project = admin_and_editor
+
+    assert await _lists(async_client, headers, project) == ([], [], [], [])
+
+
+@pytest.fixture
+async def admin_and_editor(async_client: AsyncClient, shared):
+    admin = shared["admin"]["Authorization"].removeprefix("Bearer ")
+    editor, _ = await _member(async_client, admin, "editor", None, BASE)
+    return editor, shared["project"]
+
+
+async def test_a_key_sees_what_its_owner_could(async_client: AsyncClient, shared):
+    created = await async_client.post(
+        "/api/v1/api-keys/", headers=shared["first"], json={"name": "k", "can_manage_projects": True}
+    )
+    assert created.status_code in (200, 201), created.text
+    archive, job = shared["rows"]["first"]
+
+    key = {"X-API-Key": created.json()["key"]}
+    assert await _lists(async_client, key, shared["project"]) == ([archive], [job], [archive], [job])
+
+
+async def test_a_user_cannot_unlink_anothers_archive(async_client: AsyncClient, shared, db_session):
+    mine, _ = shared["rows"]["first"]
+    theirs, _ = shared["rows"]["second"]
+
+    response = await async_client.post(
+        f"/api/v1/projects/{shared['project']}/remove-archives",
+        headers=shared["first"],
+        json={"archive_ids": [mine, theirs]},
+    )
+
+    assert response.json()["count"] == 1
+    assert await _archive_projects(db_session, mine, theirs) == [None, shared["project"]]
+
+
+async def test_auth_off_shows_every_row(async_client: AsyncClient, printer_factory, archive_factory, db_session):
+    a = await printer_factory(name="A")
+    project = (await async_client.post("/api/v1/projects/", json={"name": "P"})).json()["id"]
+    first = await archive_factory(a.id, project_id=project, status="completed")
+    second = await archive_factory(a.id, project_id=project, status="completed")
+    jobs = sorted(
+        [await _job(db_session, a.id, first.id, project, None), await _job(db_session, a.id, second.id, project, None)]
+    )
+
+    assert await _lists(async_client, {}, project) == (
+        sorted([first.id, second.id]),
+        jobs,
+        sorted([first.id, second.id]),
+        jobs,
+    )
+
+
+async def test_the_project_list_previews_only_visible_archives(async_client: AsyncClient, shared):
+    response = await async_client.get("/api/v1/projects/", headers=shared["first"])
+
+    assert response.status_code == 200, response.text
+    (project,) = [p for p in response.json() if p["id"] == shared["project"]]
+    assert [a["id"] for a in project["archives"]] == [shared["rows"]["first"][0]]
+
+
+async def test_a_bom_names_only_visible_archives(async_client: AsyncClient, shared):
+    project = shared["project"]
+    mine, theirs = shared["rows"]["first"][0], shared["rows"]["second"][0]
+    linked = await async_client.post(
+        f"/api/v1/projects/{project}/bom", headers=shared["admin"], json={"name": "b", "archive_id": theirs}
+    )
+    assert linked.status_code == 200, linked.text
+    item = linked.json()["id"]
+
+    listed = await async_client.get(f"/api/v1/projects/{project}/bom", headers=shared["first"])
+    relinked = await async_client.patch(
+        f"/api/v1/projects/{project}/bom/{item}", headers=shared["first"], json={"archive_id": theirs}
+    )
+    created = await async_client.post(
+        f"/api/v1/projects/{project}/bom", headers=shared["first"], json={"name": "c", "archive_id": theirs}
+    )
+    own = await async_client.post(
+        f"/api/v1/projects/{project}/bom", headers=shared["first"], json={"name": "d", "archive_id": mine}
+    )
+
+    assert [(b["archive_id"], b["archive_name"]) for b in listed.json()] == [(None, None)]
+    assert relinked.status_code == created.status_code == 400
+    assert (own.json()["archive_id"], own.json()["archive_name"]) == (mine, "first part")
+
+
+async def test_a_library_folder_links_only_visible_archives(
+    async_client: AsyncClient, shared, archive_factory, db_session
+):
+    from datetime import datetime
+
+    from backend.app.models.archive import PrintArchive
+
+    mine, theirs = shared["rows"]["first"][0], shared["rows"]["second"][0]
+    trashed = await db_session.get(PrintArchive, mine)
+    folder = lambda archive_id: async_client.post(  # noqa: E731
+        "/api/v1/library/folders/", headers=shared["first"], json={"name": f"f{archive_id}", "archive_id": archive_id}
+    )
+
+    other = await folder(theirs)
+    own = await folder(mine)
+    trashed.deleted_at = datetime.now()
+    await db_session.commit()
+    gone = await folder(mine)
+
+    assert other.status_code == 404
+    assert (own.json()["archive_id"], own.json()["archive_name"]) == (mine, "first part")
+    assert gone.status_code == 404
+
+
+async def test_a_key_of_a_whole_library_reader_still_sees_only_its_owners_archives(async_client: AsyncClient, shared):
+    admin = shared["admin"]["Authorization"].removeprefix("Bearer ")
+    reader, _ = await _member(
+        async_client, admin, "reader", None, [*BASE, "archives:read_own", "library:read_all", "library:update_all"]
+    )
+    theirs = shared["rows"]["second"][0]
+    folder = await async_client.post(
+        "/api/v1/library/folders/", headers=shared["admin"], json={"name": "shared", "archive_id": theirs}
+    )
+    assert folder.status_code == 200, folder.text
+    created = await async_client.post("/api/v1/api-keys/", headers=reader, json={"name": "k"})
+    assert created.status_code in (200, 201), created.text
+    key = {"X-API-Key": created.json()["key"]}
+
+    listed = await async_client.get("/api/v1/library/folders/", headers=key)
+    one = await async_client.get(f"/api/v1/library/folders/{folder.json()['id']}", headers=key)
+    by_archive = await async_client.get(f"/api/v1/library/folders/by-archive/{theirs}", headers=key)
+    bulk = await async_client.get("/api/v1/bulk/archive-folders", params={"ids": str(theirs)}, headers=key)
+    relinked = await async_client.put(
+        f"/api/v1/library/folders/{folder.json()['id']}", headers=key, json={"archive_id": theirs}
+    )
+
+    assert [(f["archive_id"], f["archive_name"]) for f in listed.json() if f["name"] == "shared"] == [(None, None)]
+    assert (one.json()["archive_id"], one.json()["archive_name"]) == (None, None)
+    assert by_archive.json() == []
+    assert bulk.json() == {str(theirs): []}
+    assert relinked.status_code == 404

backend/tests/integration/test_queue_creation_paths_c4.py

@@ -330,7 +330,7 @@ class TestTheDefaultPrintOptions:
 
 
 class TestAPipelineRun:
-    async def _run(self, db_session, test_engine, tmp_path):
+    async def _run(self, db_session, test_engine, tmp_path, owner_id=None, slices=None):
         from backend.app.api.routes.pipeline_runs import _make_orchestration_callable
         from backend.app.core.printer_scope import ALL_PRINTERS
         from backend.app.models.library import LibraryFile
@@ -352,7 +352,7 @@ class TestAPipelineRun:
         )
         db_session.add_all([source, pipeline])
         await db_session.commit()
-        run = PipelineRun(pipeline_id=pipeline.id, source_library_file_id=source.id, copies=2, created_by=None)
+        run = PipelineRun(pipeline_id=pipeline.id, source_library_file_id=source.id, copies=2, created_by=owner_id)
         db_session.add(run)
         await db_session.commit()
         db_session.add_all([PipelineJob(pipeline_run_id=run.id, copy_index=i) for i in range(2)])
@@ -361,6 +361,8 @@ class TestAPipelineRun:
         model.write_bytes(b"model")
 
         async def sliced(session, **kwargs):
+            if slices is not None:
+                slices.append(kwargs)
             out = LibraryFile(filename="pipe.gcode.3mf", file_path="library/pipe.gcode.3mf", file_type="3mf")
             out.file_size = 1
             session.add(out)
@@ -402,6 +404,19 @@ class TestAPipelineRun:
         notify.assert_awaited_once()
         assert notify.await_args.kwargs["job_name"] == "pipe ×2"
 
+    async def test_the_sliced_file_is_the_runs_owners(self, db_session, test_engine, tmp_path, announced):
+        """Also when no one resolves cloud presets: a key without the cloud scope started it."""
+        owner = User(username="runowner", password_hash="x", role="user")
+        db_session.add(owner)
+        await db_session.commit()
+        slices: list = []
+
+        items = await self._run(db_session, test_engine, tmp_path, owner_id=owner.id, slices=slices)
+
+        assert [s["current_user_id"] for s in slices] == [None]
+        assert [s["owner_id"] for s in slices] == [owner.id]
+        assert {item.created_by_id for item in items} == {owner.id}
+
 
 class TestAVirtualPrinter:
     async def test_announces_what_it_queued(self, tmp_path):

backend/tests/integration/test_queue_review_1620.py

@@ -317,10 +317,17 @@ class TestApiKeys:
         item = await _queue(async_client, _auth(setup["student"]), setup["printer"].id, setup["archive"].id)
         key = await self._key(async_client, controller, can_control_printer=True)
 
+        # Someone else's job is not theirs to start at all
         refused = await async_client.post(f"/api/v1/webhook/printer/{setup['printer'].id}/start", headers=key)
-        assert refused.status_code == 403
+        assert refused.status_code == 404
         assert await _manual_start(item["id"]) is True
 
+        # Nor their own, while it waits for review
+        own = await _queue(async_client, _auth(controller), setup["printer"].id, setup["archive"].id)
+        refused = await async_client.post(f"/api/v1/webhook/printer/{setup['printer'].id}/start", headers=key)
+        assert refused.status_code == 403
+        assert await _manual_start(own["id"]) is True
+
         staff_key = await self._key(async_client, admin, can_control_printer=True)
         started = await async_client.post(f"/api/v1/webhook/printer/{setup['printer'].id}/start", headers=staff_key)
         assert started.status_code == 200, started.text

backend/tests/integration/test_slice_archive_output_path_2832.py

@@ -137,3 +137,31 @@ class TestOrdinaryNamesAreUnchanged:
 
         assert response.name == "Benchy (re-sliced)"
         assert [p.name for p in await _written_files(archive_root) if p.suffix == ".3mf"] == ["Benchy.gcode.3mf"]
+
+
+async def test_the_new_archive_can_be_someone_other_than_the_cloud_user(
+    db_session, archive_factory, printer_factory, archive_root
+):
+    """A key without the cloud scope slices: nobody resolves cloud presets, its
+    owner owns the result."""
+    from backend.app.models.archive import PrintArchive
+    from backend.app.models.user import User
+
+    owner = User(username="archiveowner", password_hash="x", role="user")
+    db_session.add(owner)
+    await db_session.commit()
+    printer = await printer_factory()
+    source = await archive_factory(printer.id, print_name="Part", filename="Part.3mf")
+
+    with _stub_slicer():
+        response = await slice_and_persist_as_archive(
+            db_session,
+            model_bytes=b"source model",
+            model_filename="Part.3mf",
+            request=SliceRequest(printer_preset_id=1, process_preset_id=2, filament_preset_id=3),
+            source_archive=source,
+            current_user_id=None,
+            owner_id=owner.id,
+        )
+
+    assert (await db_session.get(PrintArchive, response.archive_id)).created_by_id == owner.id

backend/tests/unit/test_print_start_time_3327.py

@@ -10,8 +10,9 @@ from datetime import datetime, timezone
 
 import pytest
 
+from backend.app.core.config import settings
 from backend.app.services.archive import ArchiveService
-from backend.tests.unit.test_archive_print_plate import _three_mf, dirs, printer_id  # noqa: F401
+from backend.tests.unit.test_archive_print_plate import _three_mf, printer_id  # noqa: F401
 from backend.tests.unit.test_print_start_no_second_download_3308 import (  # noqa: F401
     FILE,
     _isolated,
@@ -22,6 +23,14 @@ from backend.tests.unit.test_print_start_no_second_download_3308 import (  # noq
 )
 
 
+@pytest.fixture
+def dirs(tmp_path, monkeypatch):
+    """The data directory; the archive directory is the print-start harness's
+    (``_isolated``), so it is patched once and put back once."""
+    monkeypatch.setattr(settings, "base_dir", tmp_path)
+    return tmp_path
+
+
 def _timed(transfer):
     """*transfer*, noting when the first download attempt began."""
     began: list[datetime] = []

backend/tests/unit/test_slice_external_folder_output.py

@@ -160,7 +160,7 @@ class TestSliceAndPersistWritesToTheMount:
             AsyncMock(return_value=(SliceResult(content, 3600, 12.5, 4200.0), False)),
         )
 
-    async def _slice_into(self, db_session, folder: LibraryFolder):
+    async def _slice_into(self, db_session, folder: LibraryFolder, **owner):
         db_session.add(folder)
         await db_session.commit()
         await db_session.refresh(folder)
@@ -174,6 +174,7 @@ class TestSliceAndPersistWritesToTheMount:
                 extra_metadata=None,
                 request=SliceRequest(printer_preset_id=1, process_preset_id=2, filament_preset_id=3),
                 current_user_id=None,
+                **owner,
             )
         file_row = await db_session.get(LibraryFile, response.library_file_id)
         return response, file_row
@@ -205,6 +206,22 @@ class TestSliceAndPersistWritesToTheMount:
         assert file_row.file_path == str(mount / "Bidoof (2).gcode.3mf")
         assert (mount / "Bidoof.gcode.3mf").read_bytes() == b"an earlier slice"
 
+    @pytest.mark.asyncio
+    async def test_the_file_can_be_someone_other_than_the_cloud_user(self, db_session, tmp_path):
+        """A pipeline run a key started: nobody resolves cloud presets, the
+        key's owner owns the result."""
+        from backend.app.models.user import User
+
+        owner = User(username="sliceowner", password_hash="x", role="user")
+        db_session.add(owner)
+        await db_session.commit()
+
+        _response, file_row = await self._slice_into(
+            db_session, LibraryFolder(name="Mine", parent_id=None, is_external=False), owner_id=owner.id
+        )
+
+        assert file_row.created_by_id == owner.id
+
     @pytest.mark.asyncio
     async def test_a_managed_folder_is_unaffected(self, db_session, tmp_path):
         folder = LibraryFolder(name="Models", parent_id=None, is_external=False)