Archives: printer scope per run (archive redesign P1b)

- A user limited to some printers sees an archive printed on one of
  them, and only their prints of it: their latest print's outcome,
  video and photos, their runs and run counts, its energy and wear
- The video and finish photos of prints on other printers stay hidden
  and are 404 on their routes; photos uploaded by hand stay visible
- Changes to the whole archive (delete, its video, its source, f3d and
  project page, its cost and outcome without a run) are refused with
  403 while it also printed on printers outside the user's scope
- Queue, reprint, slice and pipelines accept an archive seen through a
  print; compare, similar and project previews keep to archives on the
  user's own printers
- Archive updates reach the sockets of every printer it printed on,
  without naming a printer outside theirs

backend/app/api/routes/archives.py

@@ -5,6 +5,7 @@ import logging
 import re as _re
 import zipfile
 from collections import defaultdict
+from dataclasses import replace
 from datetime import date, datetime, timedelta, timezone
 from html import escape as html_escape
 from pathlib import Path
@@ -30,6 +31,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 archive_scope_clause, whole_archive_scope_clause
 from backend.app.models.archive import PrintArchive
 from backend.app.models.filament import Filament
 from backend.app.models.print_log import INCLUDE_OPEN_RUNS, PrintLogEntry
@@ -41,6 +43,13 @@ from backend.app.schemas.archive import ArchiveResponse, ArchiveSlim, ArchiveSta
 from backend.app.schemas.print_log import PrintLogResponse
 from backend.app.schemas.slicer import SliceRequest
 from backend.app.services.archive import ArchiveService
+from backend.app.services.archive_scope import (
+    ScopedView,
+    ensure_whole_archive_in_scope,
+    printed_in_scope,
+    scoped_view,
+    scoped_views,
+)
 from backend.app.services.bambu_ftp import ftps_handshake_blocked, list_files_result_async
 from backend.app.services.design_settings import overrides_from_config
 from backend.app.services.filament_requirements import annotate_rack_groups
@@ -169,10 +178,22 @@ async def _claimed_timelapse_stems(db, printer_id: int | None, exclude_archive_i
     )
 
 
-def _visible_archive_clauses(user, can_read_all: bool, printer_scope: PrinterScope) -> list:
-    """WHERE clauses for the archives a caller may see, as _ensure_archive_visible decides one by one."""
+def _visible_archive_clauses(
+    user, can_read_all: bool, printer_scope: PrinterScope, *, whole: bool = False, by_runs: bool = True
+) -> list:
+    """WHERE clauses for the archives a caller may see, as _ensure_archive_visible decides one by one.
+    ``whole``: only those every print of which they see, for a change to the
+    whole archive (``ensure_whole_archive_in_scope``). Not ``by_runs``: only
+    those on their printers, for listings that show the archive's latest
+    print as it is, without the caller's view of it (``_scoped``)."""
     clauses = []
-    if (scope_clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+    if whole:
+        scope_clause = whole_archive_scope_clause(printer_scope)
+    elif by_runs:
+        scope_clause = archive_scope_clause(printer_scope)
+    else:
+        scope_clause = printer_scope.where(PrintArchive.printer_id)
+    if scope_clause is not None:
         clauses.append(scope_clause)
     if not can_read_all:
         # Own archives only; an ownerless one needs read-all, and no user sees none.
@@ -180,7 +201,7 @@ def _visible_archive_clauses(user, can_read_all: bool, printer_scope: PrinterSco
     return clauses
 
 
-def _ensure_archive_visible(
+async def _ensure_archive_visible(
     archive: PrintArchive | None,
     user: User | ApiKeyActor | None,
     can_read_all: bool,
@@ -202,15 +223,19 @@ def _ensure_archive_visible(
         nonexistent id. Pre-GHSA fix the caller saw 200 here (the PoC vector).
       - Ownerless rows (``created_by_id is None``) require ALL — fail-closed
         per ``feedback_no_fail_open_in_auth``.
-      - An archive whose printer is outside ``printer_scope`` → 404 (#1727).
+      - An archive neither on a printer in ``printer_scope`` nor printed on
+        one (any of its runs) → 404 (#1727, archive redesign).
     """
     if not archive or archive.deleted_at is not None:
         raise HTTPException(404, "Archive not found")
     # An archive from a printer the caller can't see is as missing as the
-    # printer itself (#1727). Archives with no printer stay governed by
-    # ownership alone.
+    # printer itself (#1727), unless one of its prints ran on one of theirs:
+    # plates of one archive print on several printers. Archives with no
+    # printer stay governed by ownership alone.
     if not printer_scope.allows(archive.printer_id):
-        raise HTTPException(404, "Archive not found")
+        async with database.async_session() as db:
+            if not await printed_in_scope(db, archive, printer_scope):
+                raise HTTPException(404, "Archive not found")
     if can_read_all:
         return archive
     # Auth enabled, caller has _OWN only.
@@ -224,6 +249,50 @@ def _ensure_archive_visible(
     return archive
 
 
+async def _ensure_whole_archive(archive: PrintArchive, printer_scope: PrinterScope) -> None:
+    """For a change to the whole archive (its file, its video, the archive
+    itself): 404 when the caller can't see it, 403 when it also printed on
+    printers outside their scope (``ensure_whole_archive_in_scope``)."""
+    if printer_scope.is_unrestricted:
+        return
+    async with database.async_session() as db:
+        if not await printed_in_scope(db, archive, printer_scope):
+            raise HTTPException(404, "Archive not found")
+        await ensure_whole_archive_in_scope(db, archive, printer_scope)
+
+
+async def _media_hidden(archive: PrintArchive, printer_scope: PrinterScope, photo: str | None = None) -> bool:
+    """Whether the archive's video (or ``photo``) is of a print on a printer
+    outside the caller's scope: a camera's images go no further than the
+    printer does (#1727, archive redesign)."""
+    if printer_scope.is_unrestricted:
+        return False
+    async with database.async_session() as db:
+        view = await scoped_view(db, archive, printer_scope)
+    if view is None:
+        return False
+    if photo is not None:
+        return photo in view.hidden_photos
+    return view.hidden_timelapse is not None and view.hidden_timelapse == archive.timelapse_path
+
+
+async def _visible_photos(archive: PrintArchive, printer_scope: PrinterScope) -> list[str] | None:
+    """The archive's photos the caller may see (``_media_hidden``)."""
+    if printer_scope.is_unrestricted or not archive.photos:
+        return archive.photos
+    async with database.async_session() as db:
+        view = await scoped_view(db, archive, printer_scope)
+    if view is None:
+        return archive.photos
+    return [p for p in archive.photos if p not in view.hidden_photos] or None
+
+
+async def _ensure_media_visible(archive: PrintArchive, printer_scope: PrinterScope, photo: str | None = None) -> None:
+    """404 for media ``_media_hidden`` hides, as for media that isn't there."""
+    if await _media_hidden(archive, printer_scope, photo):
+        raise HTTPException(404, "Photo not found" if photo is not None else "Timelapse not found")
+
+
 def _validate_user_filter_permission(current_user: User | ApiKeyActor | None, created_by_id: int | None):
     """Raise 403 if created_by_id filter is used without stats:filter_by_user permission."""
     if created_by_id is None or current_user is None:
@@ -295,12 +364,17 @@ def compute_time_accuracy(archive: PrintArchive, run_aggregate: dict | None = No
     return result
 
 
-async def _load_run_aggregates(db: AsyncSession, archive_ids: list[int]) -> dict[int, dict]:
+async def _load_run_aggregates(
+    db: AsyncSession, archive_ids: list[int], printer_scope: PrinterScope | None = None
+) -> dict[int, dict]:
     """Batch-load per-archive run aggregates from PrintLogEntry.
 
     Returns ``{archive_id: {run_count, last_run_at, total_filament_actual_grams,
     successful_run_count, failed_run_count, unmeasured_run_count}}``. Archives with no logged runs are
     absent from the map; callers should treat that as zero/none.
+
+    With ``printer_scope``, only the runs on its printers count: a caller
+    limited to some printers sees only their prints (archive redesign).
     """
     from backend.app.models.print_log import PrintLogEntry
 
@@ -316,7 +390,14 @@ async def _load_run_aggregates(db: AsyncSession, archive_ids: list[int]) -> dict
             func.sum(case((PrintLogEntry.status == "failed", 1), else_=0)).label("failed"),
             func.sum(case((PrintLogEntry.duration_unmeasured.is_(True), 1), else_=0)).label("estimated"),
         )
-        .where(PrintLogEntry.archive_id.in_(archive_ids))
+        .where(
+            PrintLogEntry.archive_id.in_(archive_ids),
+            *(
+                [clause]
+                if printer_scope is not None and (clause := printer_scope.where(PrintLogEntry.printer_id)) is not None
+                else []
+            ),
+        )
         .group_by(PrintLogEntry.archive_id)
     )
     aggregates: dict[int, dict] = {}
@@ -332,6 +413,10 @@ async def _load_run_aggregates(db: AsyncSession, archive_ids: list[int]) -> dict
     return aggregates
 
 
+# What PATCH may change of an archive printed on printers outside the
+# caller's scope: the file's own notes, tags, favourite and project.
+_FILE_FIELDS = frozenset({"tags", "notes", "is_favorite", "project_id"})
+
 # What PATCH with a run_id may change: one run's outcome.
 _RUN_OUTCOME_FIELDS = frozenset({"user_verdict", "failure_reason", "status"})
 
@@ -366,6 +451,29 @@ def _with_run_outcome(data: dict, run: PrintLogEntry) -> dict:
     return data
 
 
+def _scoped(data: dict, view: ScopedView | None) -> dict:
+    """An archive response as a caller limited to some printers sees it
+    (#1727, archive redesign): their latest print of it when the archive's
+    latest is on a printer outside their scope, and without the video and
+    finish photos of prints outside it."""
+    if view is None:
+        return data
+    if view.run is not None:
+        _with_run_outcome(data, view.run)
+    # The archive's times and its energy and wear are of its latest and
+    # first prints, maybe on a printer outside the scope: the badge reads as
+    # for any archive printed more than once, and the figures are the shown
+    # print's own.
+    data["actual_time_seconds"] = data["time_accuracy"] = None
+    for name in ("energy_kwh", "energy_cost", "wear_cost"):
+        data[name] = getattr(view.shown, name) if view.shown is not None else None
+    if view.hidden_timelapse is not None and data["timelapse_path"] == view.hidden_timelapse:
+        data["timelapse_path"] = None
+    if data["photos"]:
+        data["photos"] = [p for p in data["photos"] if p not in view.hidden_photos] or None
+    return data
+
+
 def archive_to_response(
     archive: PrintArchive,
     duplicates: list[dict] | None = None,
@@ -558,7 +666,8 @@ async def list_archives(
             for sequence, (archive_id, _) in enumerate(group):
                 duplicate_meta_by_archive_id.setdefault(archive_id, (sequence, original_id, duplicate_count))
 
-    run_aggregates = await _load_run_aggregates(db, [a.id for a in archives])
+    run_aggregates = await _load_run_aggregates(db, [a.id for a in archives], printer_scope)
+    views = await scoped_views(db, archives, printer_scope)
 
     # Build response with duplicate sequence and original archive ID pre-computed
     result = []
@@ -579,12 +688,15 @@ async def list_archives(
             duplicate_sequence, original_archive_id, duplicate_count = duplicate_meta_by_archive_id[a.id]
 
         result.append(
-            archive_to_response(
-                a,
-                duplicate_count=duplicate_count,
-                duplicate_sequence=duplicate_sequence,
-                original_archive_id=original_archive_id,
-                run_aggregate=run_aggregates.get(a.id),
+            _scoped(
+                archive_to_response(
+                    a,
+                    duplicate_count=duplicate_count,
+                    duplicate_sequence=duplicate_sequence,
+                    original_archive_id=original_archive_id,
+                    run_aggregate=run_aggregates.get(a.id),
+                ),
+                views.get(a.id),
             )
         )
     return result
@@ -620,12 +732,10 @@ async def list_last_archive_per_printer(
         filters.append(PrintArchive.created_by_id == user.id)
 
     run_filters: list[ColumnElement[bool]] = [PrintLogEntry.printer_id.isnot(None)]
-    # Only printers the caller may see (#1727): the run's, and the archive's
-    # too, which the archive page and its edit go by.
+    # Only printers the caller may see (#1727): the run's. Its archive is
+    # theirs to see by that run (archive redesign).
     if (clause := printer_scope.where(PrintLogEntry.printer_id)) is not None:
         run_filters.append(clause)
-    if (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
-        run_filters.append(clause)
     ranked_runs = (
         select(
             PrintLogEntry.id,
@@ -687,10 +797,13 @@ async def list_last_archive_per_printer(
         .where(PrintArchive.id.in_(archive_ids))
     )
     archives = {a.id: a for a in result.scalars().all()}
-    run_aggregates = await _load_run_aggregates(db, list(archives))
+    run_aggregates = await _load_run_aggregates(db, list(archives), printer_scope)
+    views = await scoped_views(db, archives.values(), printer_scope)
 
     def response(archive: PrintArchive) -> dict:
-        return archive_to_response(archive, run_aggregate=run_aggregates.get(archive.id))
+        return _scoped(
+            archive_to_response(archive, run_aggregate=run_aggregates.get(archive.id)), views.get(archive.id)
+        )
 
     cards = [_with_run_outcome(response(archives[r.archive_id]), r) for r in runs if r.archive_id in archives]
     cards += [response(archives[i]) for i in without_runs if i in archives]
@@ -744,7 +857,7 @@ async def no_3mf_warning(
     ]
     if user is not None and not can_read_all:
         conditions.append(PrintArchive.created_by_id == user.id)
-    if (scope_clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+    if (scope_clause := archive_scope_clause(printer_scope)) is not None:
         conditions.append(scope_clause)
     result = await db.execute(select(PrintArchive.extra_data).where(*conditions))
     reasons: set[str] = set()
@@ -1012,7 +1125,7 @@ async def search_archives(
             query = query.where(PrintArchive.status == status)
         if own_only:
             query = query.where(PrintArchive.created_by_id == user.id)
-        if (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+        if (clause := archive_scope_clause(printer_scope)) is not None:
             query = query.where(clause)
 
         query = query.limit(limit).offset(offset)
@@ -1020,8 +1133,11 @@ async def search_archives(
         archives = result.scalars().all()
         # Load run aggregates so multi-run archives' time/accuracy badge is
         # suppressed consistently with the main list endpoint (#1608).
-        run_aggregates = await _load_run_aggregates(db, [a.id for a in archives])
-        return [archive_to_response(a, run_aggregate=run_aggregates.get(a.id)) for a in archives]
+        run_aggregates = await _load_run_aggregates(db, [a.id for a in archives], printer_scope)
+        views = await scoped_views(db, archives, printer_scope)
+        return [
+            _scoped(archive_to_response(a, run_aggregate=run_aggregates.get(a.id)), views.get(a.id)) for a in archives
+        ]
 
     if not matched_ids:
         return []
@@ -1034,7 +1150,7 @@ async def search_archives(
     )
     if own_only:
         query = query.where(PrintArchive.created_by_id == user.id)
-    if (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+    if (clause := archive_scope_clause(printer_scope)) is not None:
         query = query.where(clause)
 
     # Apply additional filters
@@ -1054,8 +1170,9 @@ async def search_archives(
 
     # Load run aggregates so multi-run archives' time/accuracy badge is
     # suppressed consistently with the main list endpoint (#1608).
-    run_aggregates = await _load_run_aggregates(db, [a.id for a in paginated])
-    return [archive_to_response(a, run_aggregate=run_aggregates.get(a.id)) for a in paginated]
+    run_aggregates = await _load_run_aggregates(db, [a.id for a in paginated], printer_scope)
+    views = await scoped_views(db, paginated, printer_scope)
+    return [_scoped(archive_to_response(a, run_aggregate=run_aggregates.get(a.id)), views.get(a.id)) for a in paginated]
 
 
 @router.post("/search/rebuild-index")
@@ -1199,10 +1316,12 @@ async def compare_archives(
             if row is None or row.deleted_at is not None or row.created_by_id != user.id:
                 raise HTTPException(404, "Archive not found")
 
-    # Every compared archive must come from a printer the caller can see (#1727)
-    if not printer_scope.is_unrestricted:
-        printer_ids = await db.execute(select(PrintArchive.printer_id).where(PrintArchive.id.in_(ids)))
-        if not all(printer_scope.allows(pid) for (pid,) in printer_ids.all()):
+    # Every compared archive must be on a printer the caller can see (#1727).
+    # Not one seen by a run only: the comparison shows each archive's latest
+    # print as it is, which may be on a printer outside their scope.
+    if (scope_clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+        seen = await db.scalar(select(func.count(PrintArchive.id)).where(PrintArchive.id.in_(ids), scope_clause))
+        if seen != len(set(ids)):
             raise HTTPException(404, "Archive not found")
 
     service = ArchiveComparisonService(db)
@@ -1760,7 +1879,7 @@ async def get_all_tags(
     tag_conditions: list[ColumnElement[bool]] = [PrintArchive.tags.isnot(None), PrintArchive.deleted_at.is_(None)]
     if user is not None and not can_read_all:
         tag_conditions.append(PrintArchive.created_by_id == user.id)
-    if (scope_clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+    if (scope_clause := archive_scope_clause(printer_scope)) is not None:
         tag_conditions.append(scope_clause)
     result = await db.execute(select(PrintArchive.tags).where(*tag_conditions))
     all_tags_rows = result.all()
@@ -1889,7 +2008,7 @@ async def get_archive(
 
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
     run = await run_of_archive(db, archive.id, run_id) if run_id is not None else None
     if run_id is not None and (run is None or not printer_scope.allows(run.printer_id)):
         raise HTTPException(404, "Run not found")
@@ -1903,9 +2022,15 @@ async def get_archive(
         makerworld_model_id=makerworld_id,
         visible=_visible_archive_clauses(user, can_read_all, printer_scope),
     )
-    run_aggregates = await _load_run_aggregates(db, [archive.id])
+    run_aggregates = await _load_run_aggregates(db, [archive.id], printer_scope)
     data = archive_to_response(archive, duplicates, run_aggregate=run_aggregates.get(archive.id))
-    return _with_run_outcome(data, run) if run is not None else data
+    view = await scoped_view(db, archive, printer_scope)
+    if view is None:
+        return _with_run_outcome(data, run) if run is not None else data
+    if run is not None:
+        # The run asked for, in the caller's scope (checked above).
+        view = replace(view, run=run, shown=run)
+    return _scoped(data, view)
 
 
 @router.get("/{archive_id}/delete-impact")
@@ -1931,7 +2056,7 @@ async def get_archive_delete_impact(
     """
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
     from backend.app.services.archive import _count_related_queue_items
 
     total, printing = await _count_related_queue_items(db, archive.id)
@@ -1958,11 +2083,13 @@ async def list_archive_runs(
     from backend.app.schemas.print_log import PrintLogEntrySchema
 
     user, can_read_all = auth_result
-    _ensure_archive_visible(await db.get(PrintArchive, archive_id), user, can_read_all, printer_scope)
+    await _ensure_archive_visible(await db.get(PrintArchive, archive_id), user, can_read_all, printer_scope)
 
+    # Only the prints on the caller's printers (#1727), as the print log.
+    scope_clause = printer_scope.where(PrintLogEntry.printer_id)
     rows = await db.execute(
         select(PrintLogEntry)
-        .where(PrintLogEntry.archive_id == archive_id)
+        .where(PrintLogEntry.archive_id == archive_id, *([scope_clause] if scope_clause is not None else []))
         .order_by(PrintLogEntry.started_at.desc().nulls_last(), PrintLogEntry.id.desc())
     )
     entries = list(rows.scalars().all())
@@ -1993,12 +2120,12 @@ async def find_similar_archives(
     from backend.app.services.archive_comparison import ArchiveComparisonService
 
     user, can_read_all = auth_result
-    _ensure_archive_visible(await db.get(PrintArchive, archive_id), user, can_read_all, printer_scope)
+    await _ensure_archive_visible(await db.get(PrintArchive, archive_id), user, can_read_all, printer_scope)
 
     service = ArchiveComparisonService(db)
     try:
         return await service.find_similar_archives(
-            archive_id, limit=limit, visible=_visible_archive_clauses(user, can_read_all, printer_scope)
+            archive_id, limit=limit, visible=_visible_archive_clauses(user, can_read_all, printer_scope, by_runs=False)
         )
     except ValueError as e:
         raise HTTPException(404, str(e))
@@ -2028,7 +2155,9 @@ async def update_archive(
         .where(PrintArchive.id == archive_id)
     )
     archive = result.scalar_one_or_none()
-    if not archive or not printer_scope.allows(archive.printer_id):
+    if not archive:
+        raise HTTPException(404, "Archive not found")
+    if not printer_scope.allows(archive.printer_id) and not await printed_in_scope(db, archive, printer_scope):
         raise HTTPException(404, "Archive not found")
 
     # Ownership check
@@ -2043,6 +2172,12 @@ async def update_archive(
     previous_status = archive.status
 
     update_payload = update_data.model_dump(exclude_unset=True)
+    # Anything but its file's own notes, tags, favourite and project, or one
+    # of the caller's runs' outcome (run_id, checked below), is the whole
+    # archive's: its outcome, cost and printer are its prints' (archive
+    # redesign).
+    if update_payload.get("run_id") is None and set(update_payload) - _FILE_FIELDS - {"run_id"}:
+        await _ensure_whole_archive(archive, printer_scope)
     # Later runs of the archive are charged from its cost and weight: a caller
     # limited to their own archives sets neither. The one exception is the
     # weight of an archive without its 3MF (#1820), which has none and cannot
@@ -2187,8 +2322,9 @@ async def update_archive(
     # Load run aggregate so the time/accuracy badge stays consistent with
     # the list / detail endpoints when the frontend re-renders the card
     # after a PATCH (#1608).
-    run_aggregates = await _load_run_aggregates(db, [archive.id]) if archive else {}
-    return archive_to_response(archive, run_aggregate=run_aggregates.get(archive.id) if archive else None)
+    run_aggregates = await _load_run_aggregates(db, [archive.id], printer_scope) if archive else {}
+    data = archive_to_response(archive, run_aggregate=run_aggregates.get(archive.id) if archive else None)
+    return _scoped(data, await scoped_view(db, archive, printer_scope)) if archive else data
 
 
 @router.post("/{archive_id}/favorite", response_model=ArchiveResponse)
@@ -2206,12 +2342,26 @@ async def toggle_favorite(
     """Toggle favorite status for an archive."""
     user, can_modify_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
 
     archive.is_favorite = not archive.is_favorite
     await db.commit()
     await db.refresh(archive)
-    return archive
+    view = await scoped_view(db, archive, printer_scope)
+    if view is None:
+        return archive
+    # As the list shows it to a caller limited to some printers.
+    from sqlalchemy.orm import selectinload
+
+    loaded = await db.scalar(
+        select(PrintArchive)
+        .options(selectinload(PrintArchive.project), selectinload(PrintArchive.created_by))
+        .where(PrintArchive.id == archive_id)
+    )
+    if loaded is None:
+        raise HTTPException(404, "Archive not found")
+    run_aggregates = await _load_run_aggregates(db, [archive_id], printer_scope)
+    return _scoped(archive_to_response(loaded, run_aggregate=run_aggregates.get(archive_id)), view)
 
 
 async def _spoolman_owns_cost(db: AsyncSession) -> bool:
@@ -2299,10 +2449,11 @@ async def recalculate_all_costs(
     from backend.app.services.print_log import mirror_archive_cost
     from backend.app.services.usage_repricing import reprice_usage
 
-    # Only the archives of the caller's printers, as the archive list shows them.
+    # Only archives every print of which is on the caller's printers: it
+    # reprices the archive and its runs (archive redesign).
     query = select(PrintArchive)
     scoped_ids = None
-    if (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+    if (clause := whole_archive_scope_clause(printer_scope)) is not None:
         query = query.where(clause)
         scoped_ids = select(PrintArchive.id).where(clause)
     result = await db.execute(query)
@@ -2423,7 +2574,7 @@ async def get_archive_duplicates(
     """Get duplicates for a specific archive."""
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     makerworld_id = archive.extra_data.get("makerworld_model_id") if archive.extra_data else None
     duplicates = await service.find_duplicates(
@@ -2504,8 +2655,10 @@ async def delete_archive(
     # Get archive first to check ownership
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
     archive = result.scalar_one_or_none()
-    if not archive or not printer_scope.allows(archive.printer_id):
+    if not archive:
         raise HTTPException(404, "Archive not found")
+    # Its prints on other printers would go with it (archive redesign).
+    await _ensure_whole_archive(archive, printer_scope)
 
     # Ownership check
     if not can_modify_all:
@@ -2570,7 +2723,7 @@ async def download_archive(
     """Download the 3MF file."""
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -2603,7 +2756,7 @@ async def download_archive_with_filename(
     """Download the 3MF file with filename in URL."""
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -2637,7 +2790,7 @@ async def create_archive_slicer_token(
 
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     token = await create_slicer_download_token("archive", archive_id)
     return {"token": token}
@@ -2698,7 +2851,7 @@ async def get_thumbnail(
     """
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
     if not archive.thumbnail_path:
         raise HTTPException(404, "Thumbnail not found")
 
@@ -2741,7 +2894,7 @@ async def get_archive_printer_media(
 
     user, can_read_all = auth_result
     async with database.async_session() as db:
-        archive = _ensure_archive_visible(
+        archive = await _ensure_archive_visible(
             await ArchiveService(db).get_archive(archive_id), user, can_read_all, printer_scope
         )
         printer = None
@@ -2752,7 +2905,7 @@ async def get_archive_printer_media(
                 claimed_timelapse_stems = await _claimed_timelapse_stems(db, archive.printer_id, archive_id)
 
     local_timelapse = None
-    if archive.timelapse_path:
+    if archive.timelapse_path and not await _media_hidden(archive, printer_scope):
         local_path = settings.base_dir / archive.timelapse_path
         if await asyncio.to_thread(local_path.is_file):
             local_timelapse = {
@@ -2762,7 +2915,8 @@ async def get_archive_printer_media(
 
     response = {
         "archive_id": archive.id,
-        "printer_id": archive.printer_id,
+        # Not a printer outside the caller's scope, seen through a run of theirs.
+        "printer_id": archive.printer_id if printer_scope.allows(archive.printer_id) else None,
         "local_timelapse": local_timelapse,
         "remote_files": [],
         "warnings": [],
@@ -2878,11 +3032,12 @@ async def create_archive_media_download_token(
 
     user, can_read_all = auth_result
     async with database.async_session() as db:
-        archive = _ensure_archive_visible(
+        archive = await _ensure_archive_visible(
             await ArchiveService(db).get_archive(archive_id), user, can_read_all, printer_scope
         )
     if not archive.timelapse_path:
         raise HTTPException(404, "Timelapse not found")
+    await _ensure_media_visible(archive, printer_scope)
     timelapse_path = settings.base_dir / archive.timelapse_path
     if not await asyncio.to_thread(timelapse_path.is_file):
         raise HTTPException(404, "Timelapse file not found")
@@ -2938,9 +3093,10 @@ async def get_timelapse(
     """
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
     if not archive.timelapse_path:
         raise HTTPException(404, "Timelapse not found")
+    await _ensure_media_visible(archive, printer_scope)
 
     timelapse_path = settings.base_dir / archive.timelapse_path
     if not timelapse_path.exists():
@@ -2980,7 +3136,8 @@ async def delete_timelapse(
     """Remove the timelapse video from an archive."""
     user, can_modify_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    await _ensure_whole_archive(archive, printer_scope)
 
     if not archive.timelapse_path:
         raise HTTPException(404, "No timelapse attached to this archive")
@@ -3041,6 +3198,7 @@ async def scan_timelapse(
             raise HTTPException(400, "Archive has no associated printer")
 
         printer_scope.ensure(archive.printer_id)
+        await _ensure_whole_archive(archive, printer_scope)
         result = await db.execute(select(Printer).where(Printer.id == archive.printer_id))
         printer = result.scalar_one_or_none()
         if not printer:
@@ -3306,6 +3464,7 @@ async def select_timelapse(
             raise HTTPException(400, "Archive has no associated printer")
 
         printer_scope.ensure(archive.printer_id)
+        await _ensure_whole_archive(archive, printer_scope)
         result = await db.execute(select(Printer).where(Printer.id == archive.printer_id))
         printer = result.scalar_one_or_none()
         if not printer:
@@ -3414,8 +3573,9 @@ async def upload_timelapse(
     """Manually upload a timelapse video to an archive."""
     service = ArchiveService(db)
     archive = await service.get_archive(archive_id)
-    if not archive or not printer_scope.allows(archive.printer_id):
+    if not archive:
         raise HTTPException(404, "Archive not found")
+    await _ensure_whole_archive(archive, printer_scope)
 
     if not file.filename or not file.filename.endswith((".mp4", ".avi", ".mkv")):
         raise HTTPException(400, "File must be a video file (.mp4, .avi, .mkv)")
@@ -3448,9 +3608,10 @@ async def get_timelapse_info(
 
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
     if not archive.timelapse_path:
         raise HTTPException(404, "Timelapse not found")
+    await _ensure_media_visible(archive, printer_scope)
 
     timelapse_path = settings.base_dir / archive.timelapse_path
     if not timelapse_path.exists():
@@ -3487,9 +3648,10 @@ async def get_timelapse_thumbnails(
 
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
     if not archive.timelapse_path:
         raise HTTPException(404, "Timelapse not found")
+    await _ensure_media_visible(archive, printer_scope)
 
     timelapse_path = settings.base_dir / archive.timelapse_path
     if not timelapse_path.exists():
@@ -3547,8 +3709,9 @@ async def process_timelapse(
 
     service = ArchiveService(db)
     archive = await service.get_archive(archive_id)
-    if not archive or not archive.timelapse_path or not printer_scope.allows(archive.printer_id):
+    if not archive or not archive.timelapse_path:
         raise HTTPException(404, "Timelapse not found")
+    await _ensure_whole_archive(archive, printer_scope)
 
     timelapse_path = settings.base_dir / archive.timelapse_path
     if not timelapse_path.exists():
@@ -3653,8 +3816,10 @@ async def _ensure_timelapse_unchanged(
 
 async def _timelapse_job_archive(archive_id: int, db: AsyncSession, printer_scope: PrinterScope) -> None:
     archive = await ArchiveService(db).get_archive(archive_id)
-    if not archive or not printer_scope.allows(archive.printer_id):
+    if not archive:
         raise HTTPException(404, "Timelapse not found")
+    # The job edits the archive's video, which is the whole archive's.
+    await _ensure_whole_archive(archive, printer_scope)
 
 
 @router.get("/{archive_id}/timelapse/job")
@@ -3715,7 +3880,7 @@ async def upload_photo(
     """Upload a photo of the printed result."""
     user, can_modify_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
 
     if not file.filename or not file.filename.lower().endswith((".jpg", ".jpeg", ".png", ".webp")):
         raise HTTPException(400, "File must be an image (.jpg, .jpeg, .png, .webp)")
@@ -3744,7 +3909,7 @@ async def upload_photo(
     await db.commit()
     await db.refresh(archive)
 
-    return {"status": "uploaded", "filename": photo_filename, "photos": archive.photos}
+    return {"status": "uploaded", "filename": photo_filename, "photos": await _visible_photos(archive, printer_scope)}
 
 
 @router.get("/{archive_id}/photos/{filename}")
@@ -3767,7 +3932,8 @@ async def get_photo(
     """
     user, can_read_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
+    await _ensure_media_visible(archive, printer_scope, filename)
 
     # Membership check first — UUID-generated names on upload mean any URL
     # filename that doesn't appear here is by definition not a real photo.
@@ -3815,7 +3981,8 @@ async def delete_photo(
     """Delete a photo."""
     user, can_modify_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    await _ensure_media_visible(archive, printer_scope, filename)
 
     if not archive.photos or filename not in archive.photos:
         raise HTTPException(404, "Photo not found")
@@ -3837,7 +4004,7 @@ async def delete_photo(
 
     await db.commit()
 
-    return {"status": "deleted", "photos": archive.photos}
+    return {"status": "deleted", "photos": await _visible_photos(archive, printer_scope)}
 
 
 # ============================================
@@ -4112,7 +4279,7 @@ async def get_qrcode(
         raise HTTPException(500, "QR code generation not available - qrcode package not installed")
 
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
 
     # Build URL to archive download
     base_url = str(request.base_url).rstrip("/")
@@ -4167,7 +4334,7 @@ async def get_archive_capabilities(
 
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -4400,7 +4567,7 @@ async def get_gcode(
     """
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -4458,7 +4625,7 @@ async def get_plate_preview(
     """
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -4686,7 +4853,7 @@ async def get_archive_plates(
 
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -5004,7 +5171,7 @@ async def get_plate_thumbnail(
     """
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -5097,7 +5264,7 @@ async def get_filament_requirements(
 
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -5271,12 +5438,12 @@ async def slice_archive(
     # though GET on that id returned 404. An API key is checked as its owner
     # (RequestActor); only auth off keeps can_read_all=True.
     can_read_all = actor is None or actor.has_permission(Permission.ARCHIVES_READ_ALL.value)
-    archive = _ensure_archive_visible(archive, actor, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(archive, actor, can_read_all, printer_scope)
     # The re-sliced copy belongs to the caller and can be queued as their own,
     # so slicing an archive needs the same right as printing it again.
     from backend.app.api.routes.print_queue import _assert_can_queue_archive
 
-    _assert_can_queue_archive(archive, actor, printer_scope)
+    await _assert_can_queue_archive(archive, actor, printer_scope)
 
     src_relative = archive.source_3mf_path or archive.file_path
     if not src_relative:
@@ -5408,7 +5575,7 @@ async def get_project_page(
 
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -5438,7 +5605,9 @@ async def update_project_page(
 
     user, can_modify_all = auth_result
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_modify_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_modify_all, printer_scope)
+    # It rewrites the archive's 3MF, the file of every print of it.
+    await _ensure_whole_archive(archive, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -5477,7 +5646,7 @@ async def get_project_image(
     from backend.app.services.archive import ProjectPageParser
 
     service = ArchiveService(db)
-    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -5565,7 +5734,8 @@ async def upload_source_3mf(
     """Upload the original source 3MF project file for an archive."""
     user, can_modify_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    await _ensure_whole_archive(archive, printer_scope)
 
     if not file.filename or not file.filename.endswith(".3mf"):
         raise HTTPException(400, "File must be a .3mf file")
@@ -5617,7 +5787,7 @@ async def download_source_3mf(
     """Download the source 3MF project file."""
     user, can_read_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
 
     if not archive.source_3mf_path:
         raise HTTPException(404, "No source 3MF attached to this archive")
@@ -5652,7 +5822,7 @@ async def download_source_3mf_for_slicer(
     """Download source 3MF with filename in URL."""
     user, can_read_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
 
     if not archive.source_3mf_path:
         raise HTTPException(404, "No source 3MF attached to this archive")
@@ -5685,7 +5855,7 @@ async def create_source_slicer_token(
 
     user, can_read_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
     if not archive.source_3mf_path:
         raise HTTPException(404, "No source 3MF attached to this archive")
 
@@ -5757,8 +5927,9 @@ async def upload_source_3mf_by_name(
             print_name = print_name[:-7]
 
     # Find matching archive - try exact match first, then fuzzy
-    # Only an archive on the caller's printers can be matched, as it is listed.
-    visible = _visible_archive_clauses(None, True, printer_scope)
+    # Only an archive every print of which is on the caller's printers: its
+    # file is the whole archive's (archive redesign).
+    visible = _visible_archive_clauses(None, True, printer_scope, whole=True)
     result = await db.execute(
         select(PrintArchive)
         .where(PrintArchive.print_name == print_name, *visible)
@@ -5838,7 +6009,8 @@ async def delete_source_3mf(
     """Delete the source 3MF project file from an archive."""
     user, can_modify_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    await _ensure_whole_archive(archive, printer_scope)
 
     if not archive.source_3mf_path:
         raise HTTPException(404, "No source 3MF attached to this archive")
@@ -5876,7 +6048,8 @@ async def upload_f3d(
     """Upload a Fusion 360 design file for an archive."""
     user, can_modify_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    await _ensure_whole_archive(archive, printer_scope)
 
     if not file.filename or not file.filename.endswith(".f3d"):
         raise HTTPException(400, "File must be a .f3d file")
@@ -5928,7 +6101,7 @@ async def download_f3d(
     """Download the Fusion 360 design file."""
     user, can_read_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all, printer_scope)
 
     if not archive.f3d_path:
         raise HTTPException(404, "No F3D file attached to this archive")
@@ -5962,7 +6135,8 @@ async def delete_f3d(
     """Delete the Fusion 360 design file from an archive."""
     user, can_modify_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    archive = await _ensure_archive_visible(result.scalar_one_or_none(), user, can_modify_all, printer_scope)
+    await _ensure_whole_archive(archive, printer_scope)
 
     if not archive.f3d_path:
         raise HTTPException(404, "No F3D file attached to this archive")

backend/app/api/routes/pipeline_runs.py

@@ -466,12 +466,12 @@ async def _resolve_source(
     assert archive_id is not None
     arc = (await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))).scalar_one_or_none()
     can_read_all = user is None or user.has_permission(Permission.ARCHIVES_READ_ALL.value)
-    arc = _ensure_archive_visible(arc, user, can_read_all, printer_scope)
+    arc = await _ensure_archive_visible(arc, user, can_read_all, printer_scope)
     # Running a pipeline on an archive prints it again: the same rules as
     # queueing it, including the right to reprint it.
     from backend.app.api.routes.print_queue import _assert_can_queue_archive
 
-    _assert_can_queue_archive(arc, user, printer_scope)
+    await _assert_can_queue_archive(arc, user, printer_scope)
     rel = arc.source_3mf_path or arc.file_path
     if not rel:
         raise HTTPException(400, "Archive has no source file to slice")

backend/app/api/routes/print_queue.py

@@ -178,7 +178,7 @@ def _extract_filament_types_from_3mf(file_path: Path, plate_id: int | None = Non
 _extract_print_time_from_3mf = extract_print_time_from_3mf
 
 
-def _assert_can_queue_archive(
+async def _assert_can_queue_archive(
     archive: PrintArchive, current_user: User | ApiKeyActor | None, printer_scope: PrinterScope
 ) -> None:
     """Gate turning *archive* into a print. Raises rather than returning a verdict.
@@ -189,8 +189,9 @@ def _assert_can_queue_archive(
 
     Three checks:
 
-    * The archive must not be deleted, and its printer must be one the caller
-      may use; otherwise it reads as missing, as the archive routes answer.
+    * The archive must not be deleted, and one the caller sees: on a printer
+      they may use, or printed on one (archive redesign); otherwise it reads
+      as missing, as the archive routes answer.
     * IDOR fix (maziggy/bambuddy-security #2): without this, a caller with
       QUEUE_CREATE could queue any user's archive even without ARCHIVES_READ on
       it — Landon's PoC enumerated this on admin's archives as operator1. Gate
@@ -206,8 +207,15 @@ def _assert_can_queue_archive(
     """
     # A deleted archive, or one from a printer the caller can't use, reads as
     # missing, like the archive routes answer for it.
-    if archive.deleted_at is not None or not printer_scope.allows(archive.printer_id):
+    if archive.deleted_at is not None:
         raise HTTPException(404, "Archive not found")
+    if not printer_scope.allows(archive.printer_id):
+        from backend.app.core import database
+        from backend.app.services.archive_scope import printed_in_scope
+
+        async with database.async_session() as db:
+            if not await printed_in_scope(db, archive, printer_scope):
+                raise HTTPException(404, "Archive not found")
     # None is auth off (for an API key too).
     if current_user is None:
         return
@@ -350,7 +358,7 @@ async def _assert_can_dispatch_batch_sources(
         archive = (await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))).scalar_one_or_none()
         # A deleted source can't be printed; dispatch will fail on it anyway.
         if archive is not None:
-            _assert_can_queue_archive(archive, current_user, printer_scope)
+            await _assert_can_queue_archive(archive, current_user, printer_scope)
 
     for library_file_id in library_file_ids:
         library_file = (
@@ -1232,7 +1240,7 @@ async def add_to_queue(
         # Missing answers like hidden (404 from the gate below).
         if not archive:
             raise HTTPException(404, "Archive not found")
-        _assert_can_queue_archive(archive, actor, printer_scope)
+        await _assert_can_queue_archive(archive, actor, printer_scope)
 
     # Validate library file exists (if provided) and get it for filament extraction
     library_file = None
@@ -1834,7 +1842,7 @@ async def create_batch(
         batch_archive = await db.get(PrintArchive, data.archive_id)
         if batch_archive is None:
             raise HTTPException(404, "Archive not found")
-        _assert_can_queue_archive(batch_archive, actor, printer_scope)
+        await _assert_can_queue_archive(batch_archive, actor, printer_scope)
     if data.library_file_id is not None:
         batch_file = (
             await db.execute(LibraryFile.active().where(LibraryFile.id == data.library_file_id))

backend/app/api/routes/projects.py

@@ -436,8 +436,10 @@ async def list_projects(
     actor: User | ApiKeyActor | None = RequestActor,
 ):
     """List all projects with basic stats."""
-    # Archive previews show only the archives the project's archive list shows
-    # the caller; the counts and totals cover every print.
+    # Archive previews show only archives on the caller's printers, each as
+    # its latest print left it (the project's archive list also shows those
+    # printed on one of theirs, as they see them); the counts and totals
+    # cover every print.
     preview_scope = visible_row_clauses(
         PrintArchive, printer_scope, actor, Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN
     )
@@ -987,7 +989,7 @@ async def list_project_archives(
         .offset(offset)
     )
     visible = visible_row_clauses(
-        PrintArchive, printer_scope, actor, Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN
+        PrintArchive, printer_scope, actor, Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN, by_runs=True
     )
     if visible is None:
         return []
@@ -995,13 +997,16 @@ async def list_project_archives(
     archives = result.scalars().all()
 
     # Import the response converter from archives module
-    from backend.app.api.routes.archives import _load_run_aggregates, archive_to_response
+    from backend.app.api.routes.archives import _load_run_aggregates, _scoped, archive_to_response
+    from backend.app.services.archive_scope import scoped_views
 
     # Load run aggregates so multi-run archives' time/accuracy badge is
     # suppressed consistently with the main archives list endpoint (#1608).
-    run_aggregates = await _load_run_aggregates(db, [a.id for a in archives])
+    run_aggregates = await _load_run_aggregates(db, [a.id for a in archives], printer_scope)
+    # As the archives list shows them to a caller limited to some printers.
+    views = await scoped_views(db, archives, printer_scope)
 
-    return [archive_to_response(a, run_aggregate=run_aggregates.get(a.id)) for a in archives]
+    return [_scoped(archive_to_response(a, run_aggregate=run_aggregates.get(a.id)), views.get(a.id)) for a in archives]
 
 
 @router.get("/{project_id}/queue")

backend/app/api/routes/webhook.py

@@ -132,7 +132,7 @@ async def webhook_add_to_queue(
     if not archive:
         raise HTTPException(status_code=404, detail="Archive not found")
     actor = await _key_actor(db, api_key)
-    _assert_can_queue_archive(archive, actor, await api_key_printer_scope(db, api_key))
+    await _assert_can_queue_archive(archive, actor, await api_key_printer_scope(db, api_key))
 
     # Verify printer exists
     result = await db.execute(select(Printer).where(Printer.id == data.printer_id))

backend/app/core/row_visibility.py

@@ -36,20 +36,74 @@ def visible_row_clauses(
     actor: User | ApiKeyActor | None,
     all_permission: Permission,
     own_permission: Permission,
+    *,
+    by_runs: bool = False,
 ) -> 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``.
+    on their printers, and by ``owned_row_clauses``. ``by_runs``: archives
+    printed on one of their printers too (``archive_scope_clause``), for a
+    listing that shows them as the caller sees them (the archives routes'
+    ``_scoped``); others show an archive's latest print as it is.
     """
     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:
+    clause = (
+        archive_scope_clause(printer_scope)
+        if model is PrintArchive and by_runs
+        else printer_scope.where(model.printer_id)
+    )
+    if clause is not None:
         return [clause, *owned]
     return owned
 
 
+def archive_scope_clause(printer_scope: PrinterScope):
+    """The archives a printer scope sees (#1727), as a WHERE clause; None
+    when it sees every printer. An archive is a file printed as runs, maybe
+    on several printers (archive redesign): it is seen on its own printer
+    (the latest run's, or none) and on every printer one of its runs printed
+    on, a print running now included.
+    """
+    from sqlalchemy import or_, select
+
+    from backend.app.models.print_log import PrintLogEntry
+
+    own = printer_scope.where(PrintArchive.printer_id)
+    if own is None:
+        return None
+    # The table, not the entity: the ORM hides open runs from every read of
+    # PrintLogEntry, and a print running on the caller's printer counts.
+    runs = PrintLogEntry.__table__
+    printed_here = select(runs.c.archive_id).where(
+        runs.c.archive_id.is_not(None), runs.c.printer_id.in_(printer_scope.printer_ids or ())
+    )
+    return or_(own, PrintArchive.id.in_(printed_here))
+
+
+def whole_archive_scope_clause(printer_scope: PrinterScope):
+    """The archives every print of which the scope sees, as a WHERE clause;
+    None when it sees every printer. What changes or removes the whole
+    archive (its file, its video, the archive itself) needs this: it would
+    change other printers' prints too (archive redesign)."""
+    from sqlalchemy import and_, not_, select
+
+    from backend.app.models.print_log import PrintLogEntry
+
+    own = printer_scope.where(PrintArchive.printer_id)
+    if own is None:
+        return None
+    runs = PrintLogEntry.__table__
+    printed_elsewhere = select(runs.c.archive_id).where(
+        runs.c.archive_id.is_not(None),
+        runs.c.printer_id.is_not(None),
+        runs.c.printer_id.not_in(printer_scope.printer_ids or ()),
+    )
+    return and_(own, not_(PrintArchive.id.in_(printed_elsewhere)))
+
+
 def may_reach_row(
     created_by_id: int | None,
     actor: User | ApiKeyActor | None,

backend/app/core/websocket.py

@@ -36,24 +36,38 @@ def _may_receive(connection: WebSocket, printer_id: int | None) -> bool:
 _UNKNOWN_PRINTER = -1
 
 
-async def _archive_printer_id(archive_id: Any) -> int | None:
-    """The printer an archive belongs to (None: printer-less, seen by everyone)."""
+async def _archive_printer_ids(archive_id: Any) -> tuple[int | None, set[int]]:
+    """The printer an archive belongs to (None: printer-less, seen by everyone),
+    and the printers its prints ran on: plates of one archive print on
+    several, and each one's sockets see it (archive redesign)."""
     from sqlalchemy import select
 
     from backend.app.core.database import async_session
     from backend.app.models.archive import PrintArchive
+    from backend.app.models.print_log import PrintLogEntry
 
     if not isinstance(archive_id, int):
-        return _UNKNOWN_PRINTER
+        return _UNKNOWN_PRINTER, set()
     try:
         async with async_session() as db:
             row = (await db.execute(select(PrintArchive.printer_id).where(PrintArchive.id == archive_id))).first()
+            # The table: the ORM leaves printing runs out, and they count.
+            runs = PrintLogEntry.__table__
+            printed_on = set(
+                (
+                    await db.scalars(
+                        select(runs.c.printer_id)
+                        .where(runs.c.archive_id == archive_id, runs.c.printer_id.is_not(None))
+                        .distinct()
+                    )
+                ).all()
+            )
     except Exception:
         logger.exception("Could not look up the printer of archive %s for a live update", archive_id)
-        return _UNKNOWN_PRINTER
+        return _UNKNOWN_PRINTER, set()
     if row is None:
-        return _UNKNOWN_PRINTER
-    return row[0]
+        return _UNKNOWN_PRINTER, set()
+    return row[0], printed_on
 
 
 class ConnectionManager:
@@ -75,20 +89,34 @@ class ConnectionManager:
             if websocket in self.active_connections:
                 self.active_connections.remove(websocket)
 
-    async def broadcast(self, message: dict[str, Any]):
-        """Broadcast a message to all connected clients."""
+    async def broadcast(
+        self,
+        message: dict[str, Any],
+        *,
+        also_printers: set[int] | None = None,
+        also_message: dict[str, Any] | None = None,
+    ):
+        """Broadcast a message to all connected clients. ``also_printers``:
+        printers besides the message's whose sockets receive it too, as
+        ``also_message`` when given (one that doesn't name the message's
+        printer, which may be outside their scope)."""
         if not self.active_connections:
             return
 
         data = json.dumps(message)
+        also_data = json.dumps(also_message) if also_message is not None else data
         printer_id = _message_printer_id(message)
         async with self._lock:
             disconnected = []
             for connection in self.active_connections:
-                if not _may_receive(connection, printer_id):
+                if _may_receive(connection, printer_id):
+                    text = data
+                elif any(_may_receive(connection, other) for other in also_printers or ()):
+                    text = also_data
+                else:
                     continue
                 try:
-                    await connection.send_text(data)
+                    await connection.send_text(text)
                 except Exception:
                     disconnected.append(connection)
 
@@ -317,10 +345,15 @@ class ConnectionManager:
         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.
         """
-        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}})
+        printer_id, printed_on = await _archive_printer_ids(archive.get("id"))
+        if isinstance(archive.get("printer_id"), int):
+            printer_id = archive["printer_id"]
+        await self.broadcast(
+            {"type": "archive_updated", "printer_id": printer_id, "data": {"printer_id": printer_id}},
+            also_printers=printed_on,
+            # Sockets that see one of its prints only: not the archive's printer.
+            also_message={"type": "archive_updated", "data": {}},
+        )
 
     async def send_queue_item_uploading(
         self,

backend/app/services/archive.py

@@ -1697,7 +1697,9 @@ class ArchiveService:
             query = query.where(PrintArchive.created_by_id == visible_to_user_id)
 
         # Only archives from printers the caller may see (#1727)
-        if printer_scope is not None and (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+        from backend.app.core.row_visibility import archive_scope_clause
+
+        if printer_scope is not None and (clause := archive_scope_clause(printer_scope)) is not None:
             query = query.where(clause)
 
         query = query.limit(limit).offset(offset)

backend/app/services/archive_scope.py

@@ -0,0 +1,144 @@
+"""What of an archive a printer-scoped caller sees and may change (#1727, archive redesign).
+
+An archive is a file; each print of it is a run, and plates of one archive
+can print on several printers. A caller limited to some printers sees an
+archive when it, or any of its runs, is on one of theirs, and then only
+their prints of it: the outcome, video and finish photos of a run on a
+printer they can't see stay hidden. What changes the whole archive needs
+every printer it printed on.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+
+from fastapi import HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.core.printer_scope import PrinterScope
+from backend.app.models.print_log import INCLUDE_OPEN_RUNS, PrintLogEntry
+
+# Finish photos are named so (main._background_finish_photo, camera); a
+# photo uploaded by hand is not, nor one carried over from the library.
+_FINISH_PHOTO_PREFIX = "finish_"
+
+
+@dataclass(frozen=True)
+class ScopedView:
+    """What a scoped caller sees of one archive beyond what it holds.
+
+    ``run`` is their latest print of it, shown instead of the archive's
+    latest when that one is on a printer outside their scope. The photos
+    and the video of prints outside it are left out.
+    """
+
+    run: PrintLogEntry | None = None
+    # The print whose figures (energy, wear) it shows: ``run``, or the
+    # archive's latest when that is in the scope.
+    shown: PrintLogEntry | None = None
+    hidden_photos: frozenset[str] = field(default_factory=frozenset)
+    # The archive's video, when it is of a print outside the scope.
+    hidden_timelapse: str | None = None
+
+
+async def runs_by_archive(db: AsyncSession, archive_ids: list[int]) -> dict[int, list[PrintLogEntry]]:
+    """Every run of each archive, a printing one included."""
+    if not archive_ids:
+        return {}
+    rows = await db.scalars(
+        select(PrintLogEntry)
+        .where(PrintLogEntry.archive_id.in_(archive_ids))
+        .execution_options(**{INCLUDE_OPEN_RUNS: True})
+    )
+    runs: dict[int, list[PrintLogEntry]] = {}
+    for run in rows.all():
+        if run.archive_id is not None:
+            runs.setdefault(run.archive_id, []).append(run)
+    return runs
+
+
+def _on_printer_in(scope: PrinterScope, run: PrintLogEntry) -> bool:
+    """A print on a printer of the scope; one whose printer was deleted is no one's."""
+    return run.printer_id is not None and scope.allows(run.printer_id)
+
+
+def view_of(archive, runs: list[PrintLogEntry], scope: PrinterScope) -> ScopedView | None:
+    """What ``scope`` sees of ``archive`` given its ``runs``; None when it sees all of it."""
+    if scope.is_unrestricted:
+        return None
+    inside = [r for r in runs if scope.allows(r.printer_id)]
+    outside = [r for r in runs if not scope.allows(r.printer_id)]
+    archive_inside = scope.allows(archive.printer_id)
+    if archive_inside and not outside:
+        return None
+
+    shown = {p for r in inside for p in (r.photos or [])}
+    hidden = {p for r in outside for p in (r.photos or []) if p not in shown}
+    if not archive_inside:
+        # A finish photo on no run's list is of a print logged before runs
+        # kept their own: the archive's latest, here on a printer outside.
+        on_runs = {p for r in runs for p in (r.photos or [])}
+        hidden |= {p for p in archive.photos or [] if p not in on_runs and p.startswith(_FINISH_PHOTO_PREFIX)}
+
+    hidden_timelapse = None
+    if archive.timelapse_path:
+        holders = [r for r in runs if r.timelapse_path == archive.timelapse_path]
+        if not any(scope.allows(r.printer_id) for r in holders) if holders else not archive_inside:
+            hidden_timelapse = archive.timelapse_path
+
+    def latest(of: list[PrintLogEntry]) -> PrintLogEntry:
+        return max(of, key=lambda r: (r.started_at is not None, r.started_at or r.created_at or datetime.min, r.id))
+
+    run = shown = None
+    if not archive_inside:
+        theirs = [r for r in runs if _on_printer_in(scope, r)]
+        if theirs:
+            run = shown = latest(theirs)
+    elif runs and scope.allows((newest := latest(runs)).printer_id):
+        shown = newest
+    return ScopedView(run=run, shown=shown, hidden_photos=frozenset(hidden), hidden_timelapse=hidden_timelapse)
+
+
+async def scoped_views(db: AsyncSession, archives, scope: PrinterScope) -> dict[int, ScopedView]:
+    """``view_of`` for each archive that ``scope`` doesn't see whole."""
+    if scope.is_unrestricted:
+        return {}
+    archives = list(archives)
+    runs = await runs_by_archive(db, [a.id for a in archives])
+    views = {}
+    for archive in archives:
+        view = view_of(archive, runs.get(archive.id, []), scope)
+        if view is not None:
+            views[archive.id] = view
+    return views
+
+
+async def scoped_view(db: AsyncSession, archive, scope: PrinterScope) -> ScopedView | None:
+    return (await scoped_views(db, [archive], scope)).get(archive.id)
+
+
+async def printed_in_scope(db: AsyncSession, archive, scope: PrinterScope) -> bool:
+    """Whether ``scope`` sees ``archive``: on its own printer, or a run on one of theirs."""
+    if scope.allows(archive.printer_id):
+        return True
+    runs = await runs_by_archive(db, [archive.id])
+    return any(_on_printer_in(scope, r) for r in runs.get(archive.id, []))
+
+
+async def ensure_whole_archive_in_scope(db: AsyncSession, archive, scope: PrinterScope) -> None:
+    """Refuse a change to the whole archive (its file, its video, the archive
+    itself) unless ``scope`` sees every printer it printed on: it would change
+    or remove those prints too. The caller sees the archive, so this says why
+    (403) rather than hiding it."""
+    if scope.is_unrestricted:
+        return
+    runs = (await runs_by_archive(db, [archive.id])).get(archive.id, [])
+    if scope.allows(archive.printer_id) and all(scope.allows(r.printer_id) for r in runs):
+        return
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail="This archive was also printed on printers outside your access, so only someone "
+        "who can see all of them can change or remove it",
+    )

backend/app/services/export.py

@@ -127,7 +127,9 @@ class ExportService:
             query = query.where(PrintArchive.created_at <= date_to)
         if visible_to_user_id is not None:
             query = query.where(PrintArchive.created_by_id == visible_to_user_id)
-        if printer_scope is not None and (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+        from backend.app.core.row_visibility import archive_scope_clause
+
+        if printer_scope is not None and (clause := archive_scope_clause(printer_scope)) is not None:
             query = query.where(clause)
         if search:
             like_pattern = f"%{search}%"

backend/tests/integration/test_archive_scope_per_run.py

@@ -0,0 +1,427 @@
+"""A caller limited to some printers sees an archive by its prints (#1727, archive redesign).
+
+Plate 1 of one archive printed on printer A, then plate 2 on B: the archive
+names B. A user of A sees the archive by plate 1 and only plate 1 of it: its
+outcome, video and finish photos. What changes the whole archive is refused
+to them, as it would change B's print too.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+
+import pytest
+
+from backend.app.core.config import settings as app_settings
+from backend.app.models.archive import PrintArchive
+from backend.app.models.print_log import RUN_OPEN, PrintLogEntry
+from backend.tests.integration.test_archives_last_per_printer import _admin_token, _auth, _user
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+T1 = datetime(2026, 10, 30, 9)
+T2 = T1 + timedelta(hours=1)
+DIR = "archives/scoped"
+PERMISSIONS = ["archives:read_all", "archives:update_all", "archives:delete_all", "printers:read", "projects:read"]
+
+
+@pytest.fixture
+def files(tmp_path, monkeypatch):
+    monkeypatch.setattr(app_settings, "base_dir", tmp_path)
+    monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
+    (tmp_path / DIR / "photos").mkdir(parents=True)
+    for name in ("a.mp4", "b.mp4", "photos/finish_a.jpg", "photos/finish_b.jpg", "photos/0a1b2c3d.jpg"):
+        (tmp_path / DIR / name).write_bytes(b"x")
+    return tmp_path
+
+
+@pytest.fixture
+async def plates(async_client, db_session, printer_factory, files):
+    """Plate 1 completed on A with its video and photo, then plate 2 failed
+    on B with its own; one photo uploaded by hand. Users scoped to A, to B
+    and to a printer C that printed nothing."""
+    a, b, c = await printer_factory(name="A"), await printer_factory(name="B"), await printer_factory(name="C")
+    archive = PrintArchive(
+        printer_id=b.id,
+        filename="two.gcode.3mf",
+        file_path=f"{DIR}/two.gcode.3mf",
+        file_size=1,
+        print_name="two",
+        status="failed",
+        started_at=T2,
+        timelapse_path=f"{DIR}/b.mp4",
+        photos=["finish_a.jpg", "finish_b.jpg", "0a1b2c3d.jpg"],
+    )
+    db_session.add(archive)
+    await db_session.flush()
+    first = PrintLogEntry(
+        archive_id=archive.id,
+        printer_id=a.id,
+        plate_id=1,
+        status="completed",
+        started_at=T1,
+        photos=["finish_a.jpg"],
+        timelapse_path=f"{DIR}/a.mp4",
+    )
+    second = PrintLogEntry(
+        archive_id=archive.id,
+        printer_id=b.id,
+        plate_id=2,
+        status="failed",
+        started_at=T2,
+        photos=["finish_b.jpg"],
+        timelapse_path=f"{DIR}/b.mp4",
+    )
+    db_session.add_all([first, second])
+    await db_session.commit()
+    admin = await _admin_token(async_client)
+    users = {}
+    for name, printer in (("of_a", a), ("of_b", b), ("of_c", c)):
+        jwt, _ = await _user(async_client, admin, name, PERMISSIONS, printer_ids=[printer.id])
+        users[name] = _auth(jwt)
+    return SimpleNamespace(
+        archive=archive.id, a=a.id, b=b.id, c=c.id, first=first.id, second=second.id, admin=_auth(admin), **users
+    )
+
+
+def _url(plates, path=""):
+    return f"/api/v1/archives/{plates.archive}{path}"
+
+
+async def _row(db_session, model, row_id):
+    db_session.expire_all()
+    return await db_session.get(model, row_id)
+
+
+class TestSeeing:
+    async def test_a_user_of_a_sees_the_archive_by_its_plate(self, async_client, plates):
+        listed = (await async_client.get("/api/v1/archives/", headers=plates.of_a)).json()
+
+        assert [a["id"] for a in listed] == [plates.archive]
+        shown = listed[0]
+        assert (shown["printer_id"], shown["status"], shown["run_id"]) == (plates.a, "completed", plates.first)
+        assert shown["timelapse_path"] == f"{DIR}/a.mp4"
+        assert shown["photos"] == ["finish_a.jpg"]
+        assert shown["run_count"] == 1
+
+    async def test_the_archive_page_shows_their_plate(self, async_client, plates):
+        response = await async_client.get(_url(plates), headers=plates.of_a)
+
+        assert response.status_code == 200, response.text
+        body = response.json()
+        assert (body["printer_id"], body["status"], body["run_id"]) == (plates.a, "completed", plates.first)
+
+    async def test_the_other_plate_is_not_theirs(self, async_client, plates):
+        response = await async_client.get(_url(plates, f"?run_id={plates.second}"), headers=plates.of_a)
+
+        assert response.status_code == 404
+
+    async def test_its_print_log_lists_their_plate_only(self, async_client, plates):
+        runs = (await async_client.get(_url(plates, "/runs"), headers=plates.of_a)).json()
+
+        assert [r["id"] for r in runs["items"]] == [plates.first]
+
+    async def test_a_user_of_b_sees_its_own_and_the_uploads(self, async_client, plates):
+        body = (await async_client.get(_url(plates), headers=plates.of_b)).json()
+
+        assert (body["printer_id"], body["status"]) == (plates.b, "failed")
+        assert body["timelapse_path"] == f"{DIR}/b.mp4"
+        assert body["photos"] == ["finish_b.jpg", "0a1b2c3d.jpg"]
+
+    async def test_a_printer_it_never_printed_on_sees_nothing(self, async_client, plates):
+        assert (await async_client.get("/api/v1/archives/", headers=plates.of_c)).json() == []
+        assert (await async_client.get(_url(plates), headers=plates.of_c)).status_code == 404
+
+    async def test_a_plate_printing_now_counts(self, async_client, db_session, plates):
+        db_session.add(
+            PrintLogEntry(
+                archive_id=plates.archive, printer_id=plates.c, status=RUN_OPEN, started_at=T2 + timedelta(hours=1)
+            )
+        )
+        await db_session.commit()
+
+        listed = (await async_client.get("/api/v1/archives/", headers=plates.of_c)).json()
+        assert [a["id"] for a in listed] == [plates.archive]
+        assert (await async_client.get(_url(plates), headers=plates.of_c)).status_code == 200
+
+    async def test_search_finds_it_by_their_plate(self, async_client, plates):
+        response = await async_client.get("/api/v1/archives/search?q=two", headers=plates.of_a)
+
+        assert response.status_code == 200, response.text
+        assert [a["id"] for a in response.json()] == [plates.archive]
+        assert response.json()[0]["photos"] == ["finish_a.jpg"]
+
+    async def test_the_admin_sees_it_whole(self, async_client, plates):
+        body = (await async_client.get(_url(plates), headers=plates.admin)).json()
+
+        assert (body["printer_id"], body["run_count"]) == (plates.b, 2)
+        assert body["photos"] == ["finish_a.jpg", "finish_b.jpg", "0a1b2c3d.jpg"]
+
+
+class TestMedia:
+    async def test_the_other_plates_video_is_not_theirs(self, async_client, plates):
+        assert (await async_client.get(_url(plates, "/timelapse"), headers=plates.of_a)).status_code == 404
+        assert (await async_client.get(_url(plates, "/timelapse"), headers=plates.of_b)).status_code == 200
+
+    async def test_the_other_plates_photo_is_not_theirs(self, async_client, plates):
+        assert (await async_client.get(_url(plates, "/photos/finish_b.jpg"), headers=plates.of_a)).status_code == 404
+        assert (await async_client.get(_url(plates, "/photos/finish_a.jpg"), headers=plates.of_a)).status_code == 200
+        assert (await async_client.get(_url(plates, "/photos/0a1b2c3d.jpg"), headers=plates.of_a)).status_code == 200
+
+    async def test_nor_theirs_to_delete(self, async_client, db_session, plates):
+        response = await async_client.delete(_url(plates, "/photos/finish_b.jpg"), headers=plates.of_a)
+
+        assert response.status_code == 404
+        assert "finish_b.jpg" in (await _row(db_session, PrintArchive, plates.archive)).photos
+
+    async def test_a_finish_photo_on_no_runs_list_is_the_archives_latest(self, async_client, db_session, plates):
+        """Logged before runs kept their own photos: the archive's, here B's."""
+        archive = await _row(db_session, PrintArchive, plates.archive)
+        archive.photos = [*archive.photos, "finish_old.jpg"]
+        await db_session.commit()
+
+        assert "finish_old.jpg" not in (await async_client.get(_url(plates), headers=plates.of_a)).json()["photos"]
+        assert "finish_old.jpg" in (await async_client.get(_url(plates), headers=plates.of_b)).json()["photos"]
+
+    async def test_a_plate_logged_before_runs_kept_media_shows_none_of_the_others(
+        self, async_client, db_session, files, plates
+    ):
+        """Its run has no video or photo list of its own: the archive's are
+        B's latest, and only the photo uploaded by hand is everyone's."""
+        first = await _row(db_session, PrintLogEntry, plates.first)
+        first.photos, first.timelapse_path = None, None
+        await db_session.commit()
+
+        body = (await async_client.get(_url(plates), headers=plates.of_a)).json()
+        assert (body["run_id"], body["timelapse_path"]) == (plates.first, None)
+        assert body["photos"] == ["0a1b2c3d.jpg"]
+        assert (await async_client.get(_url(plates, "/photos/finish_a.jpg"), headers=plates.of_a)).status_code == 404
+
+
+class TestWholeArchive:
+    async def test_deleting_it_is_refused(self, async_client, db_session, plates):
+        response = await async_client.delete(_url(plates), headers=plates.of_a)
+
+        assert response.status_code == 403, response.text
+        assert (await _row(db_session, PrintArchive, plates.archive)).deleted_at is None
+
+    async def test_also_to_the_user_of_its_own_printer(self, async_client, plates):
+        """B is the archive's printer; plate 1 printed on A."""
+        assert (await async_client.delete(_url(plates), headers=plates.of_b)).status_code == 403
+
+    async def test_deleting_its_video_is_refused(self, async_client, db_session, plates):
+        response = await async_client.delete(_url(plates, "/timelapse"), headers=plates.of_b)
+
+        assert response.status_code == 403
+        assert (await _row(db_session, PrintArchive, plates.archive)).timelapse_path == f"{DIR}/b.mp4"
+
+    async def test_its_notes_are_theirs_to_edit(self, async_client, db_session, plates):
+        response = await async_client.patch(_url(plates), headers=plates.of_a, json={"notes": "nice", "tags": "x"})
+
+        assert response.status_code == 200, response.text
+        assert (await _row(db_session, PrintArchive, plates.archive)).notes == "nice"
+
+    async def test_its_cost_is_not(self, async_client, db_session, plates):
+        response = await async_client.patch(_url(plates), headers=plates.of_a, json={"cost": 9.0})
+
+        assert response.status_code == 403
+        assert (await _row(db_session, PrintArchive, plates.archive)).cost is None
+
+    async def test_nor_its_outcome_without_their_run(self, async_client, db_session, plates):
+        response = await async_client.patch(_url(plates), headers=plates.of_a, json={"user_verdict": "good"})
+
+        assert response.status_code == 403
+
+    async def test_their_plates_outcome_is(self, async_client, db_session, plates):
+        response = await async_client.patch(
+            _url(plates), headers=plates.of_a, json={"user_verdict": "good", "run_id": plates.first}
+        )
+
+        assert response.status_code == 200, response.text
+        assert (await _row(db_session, PrintLogEntry, plates.first)).user_verdict == "good"
+
+    async def test_the_admin_may_delete_it(self, async_client, db_session, plates):
+        response = await async_client.delete(_url(plates), headers=plates.admin)
+
+        assert response.status_code == 200, response.text
+
+    @pytest.mark.parametrize(
+        ("method", "path", "kwargs"),
+        [
+            ("post", "/timelapse/select?filename=v.mp4", {}),
+            ("post", "/timelapse/upload", {"files": {"file": ("v.mp4", b"x", "video/mp4")}}),
+            ("post", "/timelapse/process", {"data": {"speed": "2"}}),
+            ("delete", "/timelapse/job", {}),
+            ("post", "/source", {"files": {"file": ("s.3mf", b"x", "application/octet-stream")}}),
+            ("delete", "/source", {}),
+            ("post", "/f3d", {"files": {"file": ("d.f3d", b"x", "application/octet-stream")}}),
+            ("delete", "/f3d", {}),
+        ],
+    )
+    async def test_its_file_and_video_are_not_theirs_to_change(self, async_client, plates, method, path, kwargs):
+        """B's user: the archive's own printer, but plate 1 printed on A."""
+        response = await getattr(async_client, method)(_url(plates, path), headers=plates.of_b, **kwargs)
+
+        assert response.status_code == 403, response.text
+
+    async def test_scanning_for_its_video_is_not_theirs(self, async_client, db_session, plates):
+        """A scan attaches a video to the archive: only without one."""
+        archive = await _row(db_session, PrintArchive, plates.archive)
+        archive.timelapse_path = None
+        await db_session.commit()
+
+        response = await async_client.post(_url(plates, "/timelapse/scan"), headers=plates.of_b)
+
+        assert response.status_code == 403, response.text
+
+
+class TestTheOtherPlatesVideo:
+    @pytest.mark.parametrize(
+        ("method", "path"),
+        [
+            ("get", "/timelapse/info"),
+            ("get", "/timelapse/thumbnails"),
+            ("post", "/media-download-token"),
+        ],
+    )
+    async def test_is_missing_for_them(self, async_client, plates, method, path):
+        response = await getattr(async_client, method)(_url(plates, path), headers=plates.of_a)
+
+        assert response.status_code == 404, response.text
+
+    async def test_is_not_offered_as_attached(self, async_client, plates):
+        response = await async_client.get(_url(plates, "/printer-media"), headers=plates.of_a)
+
+        assert response.status_code == 200, response.text
+        assert response.json()["local_timelapse"] is None
+
+
+class TestReviewFixes:
+    async def test_a_null_run_id_is_no_run(self, async_client, db_session, plates):
+        response = await async_client.patch(
+            _url(plates), headers=plates.of_a, json={"run_id": None, "status": "completed", "failure_reason": "x"}
+        )
+
+        assert response.status_code == 403, response.text
+        assert (await _row(db_session, PrintArchive, plates.archive)).status == "failed"
+
+    async def test_its_project_page_is_its_file(self, async_client, plates):
+        response = await async_client.patch(_url(plates, "/project-page"), headers=plates.of_a, json={"title": "x"})
+
+        assert response.status_code == 403, response.text
+
+    async def test_compare_and_similar_leave_it_out(self, async_client, db_session, plates):
+        """Both show each archive's latest print as it is: B's."""
+        other = PrintArchive(
+            printer_id=plates.a, filename="two.gcode.3mf", file_path=f"{DIR}/o.3mf", file_size=1, print_name="two"
+        )
+        db_session.add(other)
+        await db_session.commit()
+        other_id = other.id
+
+        compare = await async_client.get(
+            f"/api/v1/archives/compare?archive_ids={plates.archive},{other_id}", headers=plates.of_a
+        )
+        assert compare.status_code == 404, compare.text
+        similar = (await async_client.get(f"/api/v1/archives/{other_id}/similar", headers=plates.of_a)).json()
+        assert plates.archive not in [s["archive"]["id"] for s in similar]
+
+    async def test_the_project_previews_leave_it_out(self, async_client, db_session, plates):
+        from backend.app.models.project import Project
+
+        project = Project(name="P")
+        db_session.add(project)
+        await db_session.flush()
+        (await db_session.get(PrintArchive, plates.archive)).project_id = project.id
+        await db_session.commit()
+
+        projects = (await async_client.get("/api/v1/projects/", headers=plates.of_a)).json()
+        assert [p.get("archives", []) for p in projects] == [[]]
+
+    async def test_its_photo_responses_show_theirs(self, async_client, plates):
+        uploaded = await async_client.post(
+            _url(plates, "/photos"), headers=plates.of_a, files={"file": ("p.jpg", b"\xff\xd8\xff", "image/jpeg")}
+        )
+
+        assert uploaded.status_code == 200, uploaded.text
+        assert "finish_b.jpg" not in uploaded.json()["photos"]
+        deleted = await async_client.delete(_url(plates, "/photos/0a1b2c3d.jpg"), headers=plates.of_a)
+        assert deleted.status_code == 200, deleted.text
+        assert "finish_b.jpg" not in deleted.json()["photos"]
+
+    async def test_printer_media_does_not_name_the_other_printer(self, async_client, plates):
+        response = await async_client.get(_url(plates, "/printer-media"), headers=plates.of_a)
+
+        assert response.json()["printer_id"] is None
+
+    async def test_its_energy_and_time_are_their_prints(self, async_client, db_session, plates):
+        archive = await _row(db_session, PrintArchive, plates.archive)
+        archive.energy_kwh, archive.wear_cost = 5.0, 2.0
+        archive.status, archive.completed_at, archive.print_time_seconds = "completed", T2 + timedelta(hours=1), 3600
+        first = await db_session.get(PrintLogEntry, plates.first)
+        first.energy_kwh, first.wear_cost = 1.0, 0.5
+        await db_session.commit()
+
+        body = (await async_client.get(_url(plates), headers=plates.of_a)).json()
+        assert (body["energy_kwh"], body["wear_cost"]) == (1.0, 0.5)
+        assert body["actual_time_seconds"] is None
+        whole = (await async_client.get(_url(plates), headers=plates.admin)).json()
+        assert (whole["energy_kwh"], whole["wear_cost"]) == (5.0, 2.0)
+
+
+async def test_queueing_it_is_theirs(db_session, plates, test_engine):
+    """It prints on their printer (archive redesign)."""
+    from unittest.mock import patch
+
+    from fastapi import HTTPException
+    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+    from backend.app.api.routes.print_queue import _assert_can_queue_archive
+    from backend.app.core.printer_scope import PrinterScope
+
+    archive = await db_session.get(PrintArchive, plates.archive)
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    with patch("backend.app.core.database.async_session", maker):
+        await _assert_can_queue_archive(archive, None, PrinterScope(frozenset({plates.a})))
+        with pytest.raises(HTTPException) as refused:
+            await _assert_can_queue_archive(archive, None, PrinterScope(frozenset({plates.c})))
+    assert refused.value.status_code == 404
+
+
+class TestLiveUpdates:
+    async def test_reach_the_printers_its_prints_ran_on(self, db_session, plates, test_engine):
+        from unittest.mock import patch
+
+        from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+        from backend.app.core import websocket
+
+        db_session.add(
+            PrintLogEntry(
+                archive_id=plates.archive, printer_id=plates.c, status=RUN_OPEN, started_at=T2 + timedelta(hours=1)
+            )
+        )
+        await db_session.commit()
+        maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+        with patch("backend.app.core.database.async_session", maker):
+            assert await websocket._archive_printer_ids(plates.archive) == (plates.b, {plates.a, plates.b, plates.c})
+
+
+async def test_the_project_list_shows_their_plate(async_client, db_session, plates):
+    from backend.app.models.project import Project
+
+    project = Project(name="P")
+    db_session.add(project)
+    await db_session.flush()
+    archive = await db_session.get(PrintArchive, plates.archive)
+    archive.project_id = project.id
+    project_id = project.id
+    await db_session.commit()
+
+    response = await async_client.get(f"/api/v1/projects/{project_id}/archives", headers=plates.of_a)
+
+    assert response.status_code == 200, response.text
+    assert [(a["id"], a["printer_id"], a["photos"]) for a in response.json()] == [
+        (plates.archive, plates.a, ["finish_a.jpg"])
+    ]

backend/tests/integration/test_verdict_per_run.py

@@ -312,9 +312,10 @@ async def test_a_run_on_a_printer_outside_the_callers_scope_is_not_theirs(async_
     assert [(c["printer_id"], c["run_id"]) for c in cards] == [(two_plates.b, two_plates.second)]
 
 
-async def test_a_card_is_not_shown_for_an_archive_outside_the_callers_scope(async_client, db_session, two_plates):
+async def test_a_card_is_shown_for_the_plate_printed_on_the_callers_printer(async_client, db_session, two_plates):
     """Scoped to printer A, whose plate it printed, while the archive names
-    B: as before runs, the archive is not theirs to see."""
+    B: the archive is theirs to see by that plate (archive redesign), and the
+    card is that plate's."""
     from backend.tests.integration.test_archives_last_per_printer import _admin_token, _auth, _user
 
     admin = await _admin_token(async_client)
@@ -323,4 +324,6 @@ async def test_a_card_is_not_shown_for_an_archive_outside_the_callers_scope(asyn
     )
 
     cards = (await async_client.get("/api/v1/archives/last-per-printer", headers=_auth(jwt))).json()
-    assert cards == []
+    assert [(c["id"], c["run_id"], c["printer_id"]) for c in cards] == [
+        (two_plates.archive, two_plates.first, two_plates.a)
+    ]

backend/tests/unit/test_archive_updated_scope.py

@@ -1,4 +1,5 @@
-"""Live archive updates reach only sockets that may see the archive's printer."""
+"""Live archive updates reach only sockets that may see the archive's printer,
+or a printer one of its prints ran on (archive redesign)."""
 
 from __future__ import annotations
 
@@ -31,13 +32,44 @@ async def test_an_update_follows_the_archive_printer(archive_printer, expected):
     sockets = {"team_a": _socket({1}), "team_b": _socket({2}), "everyone": _socket(None)}
     manager.active_connections = list(sockets.values())
 
-    with patch.object(websocket, "_archive_printer_id", new=AsyncMock(return_value=archive_printer)):
+    with patch.object(websocket, "_archive_printer_ids", new=AsyncMock(return_value=(archive_printer, set()))):
         await manager.send_archive_updated({"id": 7, "cost": 1.5})
 
     assert {name: s.send_text.await_count == 1 for name, s in sockets.items()} == expected
 
 
+async def test_an_update_reaches_the_printers_its_prints_ran_on():
+    """Plate 1 printed on 1, plate 2 on 2: the archive names 2, both see it."""
+    manager = websocket.ConnectionManager()
+    sockets = {"team_a": _socket({1}), "team_b": _socket({2}), "team_c": _socket({3})}
+    manager.active_connections = list(sockets.values())
+
+    with patch.object(websocket, "_archive_printer_ids", new=AsyncMock(return_value=(2, {1, 2}))):
+        await manager.send_archive_updated({"id": 7, "cost": 1.5})
+
+    assert {name: s.send_text.await_count == 1 for name, s in sockets.items()} == {
+        "team_a": True,
+        "team_b": True,
+        "team_c": False,
+    }
+
+
+async def test_a_socket_that_sees_only_a_print_of_it_is_not_told_its_printer():
+    """The archive names 2, outside team_a's scope (#1727)."""
+    import json
+
+    manager = websocket.ConnectionManager()
+    team_a, team_b = _socket({1}), _socket({2})
+    manager.active_connections = [team_a, team_b]
+
+    with patch.object(websocket, "_archive_printer_ids", new=AsyncMock(return_value=(2, {1, 2}))):
+        await manager.send_archive_updated({"id": 7})
+
+    assert json.loads(team_a.send_text.await_args.args[0]) == {"type": "archive_updated", "data": {}}
+    assert json.loads(team_b.send_text.await_args.args[0])["printer_id"] == 2
+
+
 async def test_an_archive_that_cannot_be_looked_up_reaches_only_unrestricted_sockets():
     with patch("backend.app.core.database.async_session", side_effect=RuntimeError("db down")):
-        assert await websocket._archive_printer_id(7) == websocket._UNKNOWN_PRINTER
-    assert await websocket._archive_printer_id(None) == websocket._UNKNOWN_PRINTER
+        assert await websocket._archive_printer_ids(7) == (websocket._UNKNOWN_PRINTER, set())
+    assert await websocket._archive_printer_ids(None) == (websocket._UNKNOWN_PRINTER, set())