Archives: merge duplicate archives of one file (archive redesign P2)

- Archives of the same file, owner and project merge into the oldest:
  their runs, queue items, batches, BOM items, spool usage, pipeline
  runs, library folders and pending verdicts move to it, and their
  photos, videos and source files move into its folder
- A print logged before runs gets a run from its archive; a video or
  one-tap link the archive alone held stays with that print
- The merged archives stay soft-deleted with an alias, their charges
  and reservations untouched; their ids keep finding the archive, also
  for a print queued on one
- An archive at work is left out, and work in progress holds its
  archive against a merge; so is a group with different source files
- GET /archives/merge-preview and POST /archives/merge, for users who
  may edit and delete all archives on every printer they printed on

backend/app/api/routes/archives.py

@@ -12,6 +12,7 @@ from pathlib import Path
 
 from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Request, UploadFile
 from fastapi.responses import FileResponse, Response
+from pydantic import BaseModel, Field
 from sqlalchemy import and_, case, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.sql.elements import ColumnElement
@@ -702,6 +703,63 @@ async def list_archives(
     return result
 
 
+@router.get("/merge-preview")
+async def preview_archive_merge(
+    db: AsyncSession = Depends(get_db, scope="function"),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_UPDATE_ALL, Permission.ARCHIVES_DELETE_ALL),
+    printer_scope: PrinterScope = RequestPrinterScope,
+):
+    """The archives that hold the same file, which "Merge duplicates" makes
+    one (archive redesign P2): same file, owner and project. The oldest stays;
+    a group with an archive at work (printing, queued, just printed) is left
+    out for now (``skipped``)."""
+    from backend.app.services.archive_merge import find_groups, merge_scope_clauses
+
+    def brief(archive: PrintArchive, runs: int) -> dict:
+        return {
+            "id": archive.id,
+            "print_name": archive.print_name,
+            "filename": archive.filename,
+            "created_at": archive.created_at,
+            "run_count": runs,
+        }
+
+    groups = await find_groups(db, merge_scope_clauses(printer_scope))
+    return {
+        "groups": [
+            {
+                "survivor": brief(g.survivor, g.run_counts.get(g.survivor.id, 0)),
+                "merged": [brief(a, g.run_counts.get(a.id, 0)) for a in g.merged],
+                "skipped": g.skipped,
+            }
+            for g in groups
+        ]
+    }
+
+
+class ArchiveMergeRequest(BaseModel):
+    # Each group by its oldest archive, the one that stays; its members are
+    # taken as they stand when the merge runs.
+    survivor_ids: list[int] = Field(min_length=1, max_length=500)
+
+
+@router.post("/merge")
+async def merge_archives(
+    body: ArchiveMergeRequest,
+    db: AsyncSession = Depends(get_db, scope="function"),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_UPDATE_ALL, Permission.ARCHIVES_DELETE_ALL),
+    printer_scope: PrinterScope = RequestPrinterScope,
+):
+    """Merge the duplicate groups of ``survivor_ids`` (see ``/merge-preview``)."""
+    from backend.app.core.websocket import ws_manager
+    from backend.app.services.archive_merge import merge, merge_scope_clauses
+
+    merged, skipped = await merge(db, list(dict.fromkeys(body.survivor_ids)), merge_scope_clauses(printer_scope))
+    for group in merged:
+        await ws_manager.send_archive_updated({"id": group["survivor_id"]})
+    return {"merged": merged, "skipped": skipped}
+
+
 @router.get("/last-per-printer", response_model=list[ArchiveResponse])
 async def list_last_archive_per_printer(
     db: AsyncSession = Depends(get_db, scope="function"),
@@ -2008,7 +2066,15 @@ async def get_archive(
 
     user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all, printer_scope)
+    found = await service.get_archive(archive_id)
+    if found is None or found.deleted_at is not None:
+        # Merged into another archive (archive redesign P2): that one, as a
+        # link or bookmark to the merged archive meant.
+        from backend.app.services.archive_merge import resolve_alias
+
+        if (into := await resolve_alias(db, archive_id)) is not None:
+            found = await service.get_archive(into)
+    archive = await _ensure_archive_visible(found, 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")

backend/app/core/database.py

@@ -359,6 +359,9 @@ NON_FK_ID_REFERENCES: dict[str, tuple[tuple[str, str], ...]] = {
         ("notification_lock_screen_widgets", "printer_id"),
     ),
     "api_keys": (("auth_ephemeral_tokens", "api_key_id"),),
+    # A merged archive's id keeps finding the archive it went into after its
+    # soft-deleted row is purged (archive redesign P2).
+    "print_archives": (("archive_aliases", "merged_archive_id"),),
 }
 
 

backend/app/core/sqlite_fk.py

@@ -42,6 +42,7 @@ APPLIED: frozenset[tuple[str, str]] = frozenset(
     {
         # print_archives
         ("active_print_spoolman", "archive_id"),
+        ("archive_aliases", "archive_id"),
         ("budget_reservations", "print_archive_id"),
         ("library_folders", "archive_id"),
         ("pending_uploads", "archived_id"),

backend/app/main.py

@@ -6767,9 +6767,16 @@ def _schedule_timelapse_scan(
     unread = printer_id in _timelapse_baselines_unread
     _timelapse_baselines_unread.discard(printer_id)
     baseline = _timelapse_baselines.pop(printer_id, None)
+    from backend.app.services import archive_work
+
+    # Held while it looks: it finds its print by the archive's id, which a
+    # merge would move away (archive redesign P2).
     spawn_background_task(
-        _scan_for_timelapse_with_retries(
-            archive_id, baseline, plate_id=plate_id, baseline_unread=unread, run_id=run_id
+        archive_work.held_while(
+            archive_id,
+            _scan_for_timelapse_with_retries(
+                archive_id, baseline, plate_id=plate_id, baseline_unread=unread, run_id=run_id
+            ),
         ),
         name=f"scan-timelapse-{archive_id}",
     )
@@ -10178,12 +10185,17 @@ async def _close_out_print(printer_id: int, data: dict):
             # silently drops the other's entry, leaving a JPEG on disk that the
             # gallery never lists.
             if timelapse_still_pending:
+                from backend.app.services import archive_work
+
                 spawn_background_task(
-                    _upgrade_finish_photo_from_timelapse(
+                    archive_work.held_while(
                         archive_id,
-                        archive_dir,
-                        rotation=getattr(printer, "camera_rotation", 0),
-                        run_id=completed_run_id,
+                        _upgrade_finish_photo_from_timelapse(
+                            archive_id,
+                            archive_dir,
+                            rotation=getattr(printer, "camera_rotation", 0),
+                            run_id=completed_run_id,
+                        ),
                     ),
                     name=f"finish-photo-upgrade-{archive_id}",
                 )
@@ -10486,7 +10498,14 @@ async def _close_out_print(printer_id: int, data: dict):
                 except Exception:
                     pass  # Best-effort timelapse session cancellation on error
 
-    spawn_background_task(_background_layer_timelapse(), name="background-layer-timelapse")
+    from backend.app.services import archive_work
+
+    # Held through the stitch, which can take longer than a merge waits after
+    # a print (archive redesign P2).
+    layer_timelapse = _background_layer_timelapse()
+    if archive_id:
+        layer_timelapse = archive_work.held_while(archive_id, layer_timelapse)
+    spawn_background_task(layer_timelapse, name="background-layer-timelapse")
 
     log_timing("All background tasks scheduled")
 

backend/app/models/__init__.py

@@ -3,6 +3,7 @@ from backend.app.models.ams_label import AmsLabel
 from backend.app.models.announcement import Announcement, AnnouncementRead
 from backend.app.models.api_key import APIKey
 from backend.app.models.archive import PrintArchive
+from backend.app.models.archive_alias import ArchiveAlias
 from backend.app.models.auth_ephemeral import AuthEphemeralToken, AuthRateLimitEvent
 from backend.app.models.color_catalog import ColorCatalogEntry
 from backend.app.models.connected_app import ConnectedApp, ConnectedAppGrant
@@ -58,6 +59,7 @@ __all__ = [
     "Printer",
     "PrinterLocation",
     "PrintArchive",
+    "ArchiveAlias",
     "Filament",
     "Settings",
     "SmartPlug",

backend/app/models/archive_alias.py

@@ -0,0 +1,25 @@
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.app.core.database import Base
+
+
+class ArchiveAlias(Base):
+    """An archive merged into another (archive redesign P2): its id still
+    finds the archive it went into, so old links, bookmarks and API calls
+    keep working. The merged row stays, soft-deleted, so the money booked
+    against it (wallet charges, budget reservations) is never rewritten.
+
+    Flattened: an archive merged on later passes its aliases to the one it
+    went into, so an id resolves in one step.
+    """
+
+    __tablename__ = "archive_aliases"
+
+    # The merged archive's id. Not a foreign key: the merged row may be purged
+    # with the other soft-deleted archives, and its id must keep resolving.
+    merged_archive_id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    archive_id: Mapped[int] = mapped_column(ForeignKey("print_archives.id", ondelete="CASCADE"), index=True)
+    merged_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

backend/app/services/archive.py

@@ -1992,8 +1992,12 @@ class ArchiveService:
 
         # For non-MP4 videos (e.g. AVI from P1S), kick off background conversion
         if not filename.lower().endswith(".mp4"):
+            from backend.app.services import archive_work
+
+            # Held from the spawn, not from when the task first runs: a merge
+            # in between would move the video out from under it.
             spawn_background_task(
-                _convert_timelapse_to_mp4(archive_id, timelapse_file),
+                archive_work.held_while(archive_id, _convert_held_timelapse(archive_id, timelapse_file)),
                 name=f"timelapse-convert-{archive_id}",
             )
 
@@ -2039,8 +2043,17 @@ async def _convert_timelapse_to_mp4(archive_id: int, source_path: Path) -> None:
     """Background task: convert non-MP4 timelapse (e.g. AVI from P1S) to MP4.
 
     Runs with low CPU priority (-threads 1, nice) so it doesn't starve
-    other processes on resource-constrained devices like Raspberry Pi.
+    other processes on resource-constrained devices like Raspberry Pi. Holds
+    the archive meanwhile (archive_work): a merge would move the video out
+    from under it.
     """
+    from backend.app.services import archive_work
+
+    await archive_work.held_while(archive_id, _convert_held_timelapse(archive_id, source_path))
+
+
+async def _convert_held_timelapse(archive_id: int, source_path: Path) -> None:
+    """``_convert_timelapse_to_mp4``'s conversion."""
     import asyncio
 
     from backend.app.core.database import async_session

backend/app/services/archive_merge.py

@@ -0,0 +1,640 @@
+"""Merge duplicate archives of one file into one (archive redesign P2).
+
+An archive is a file; each print of it is a run. Before runs, every dispatch
+of a library or project file archived it anew, so a file printed plate by
+plate has several archives. They are one when they hold the same file
+(content hash), have the same owner and are in the same project.
+
+Merging keeps the oldest archive and moves everything the others hold to it:
+their runs (one is written for an archive printed before runs were logged),
+queue items, batches, BOM items, pipeline runs, library folders, spool usage
+and pending Telegram verdicts, and their photos and videos, into its folder.
+The others stay, soft-deleted, each with an alias to the one it went into: the
+money booked against them (wallet charges, budget reservations) is never
+rewritten, and their ids keep finding the archive (``resolve_alias``).
+
+An archive at work is left out: printing, queued, tracked by Spoolman, being
+dispatched or converted in this process (services/archive_work), sliced by a
+pipeline, edited, or printed within the last half hour (its timelapse and
+finish photo may still be on their way, by its own id). So is a group whose
+archives carry different source or design files: only one can stay.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import logging
+import os
+import shutil
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from sqlalchemy import func, or_, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.core.config import settings
+from backend.app.core.printer_scope import PrinterScope
+from backend.app.models.active_print_spoolman import ActivePrintSpoolman
+from backend.app.models.archive import PrintArchive
+from backend.app.models.archive_alias import ArchiveAlias
+from backend.app.models.print_log import INCLUDE_OPEN_RUNS, RUN_OPEN, PrintLogEntry
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.services import archive_work
+from backend.app.utils.archive_paths import archive_dir, archive_photos_dir, find_archive_photo
+from backend.app.utils.local_time import utcnow_naive
+from backend.app.utils.safe_path import PathTraversalError, safe_join_under
+
+logger = logging.getLogger(__name__)
+
+# Long enough for the completion's timelapse scan and finish-photo upgrade,
+# which look for their print by its archive's id.
+_JUST_PRINTED = timedelta(minutes=30)
+# What an archive printed before runs were logged can say of its print.
+_ENDED = frozenset({"completed", "failed", "cancelled", "aborted", "stopped"})
+_PIPELINE_ACTIVE = ("queued", "slicing", "dispatching", "in_progress")
+# The file's own extras, of which the archive keeps one each.
+_FILE_EXTRAS = ("source_3mf_path", "f3d_path")
+
+SKIPPED_BUSY = "busy"
+SKIPPED_CONFLICTING_FILES = "conflicting_files"
+SKIPPED_FAILED = "failed"
+
+
+@dataclass
+class MergeGroup:
+    survivor: PrintArchive
+    merged: list[PrintArchive]
+    run_counts: dict[int, int] = field(default_factory=dict)
+    skipped: str | None = None
+
+
+async def resolve_alias(db: AsyncSession, archive_id: int) -> int | None:
+    """The archive a merged one went into, or None."""
+    return await db.scalar(select(ArchiveAlias.archive_id).where(ArchiveAlias.merged_archive_id == archive_id))
+
+
+def _key(archive: PrintArchive) -> tuple:
+    return (archive.content_hash, archive.created_by_id, archive.project_id)
+
+
+async def find_groups(db: AsyncSession, visible: list) -> list[MergeGroup]:
+    """The duplicate groups among the archives ``visible`` (WHERE clauses)
+    admits, oldest first, each with why it is left out (``skipped``)."""
+    keys = (
+        select(PrintArchive.content_hash, PrintArchive.created_by_id, PrintArchive.project_id)
+        .where(PrintArchive.content_hash.is_not(None), PrintArchive.deleted_at.is_(None), *visible)
+        .group_by(PrintArchive.content_hash, PrintArchive.created_by_id, PrintArchive.project_id)
+        .having(func.count(PrintArchive.id) > 1)
+    )
+    hashes = {row.content_hash for row in (await db.execute(keys)).all()}
+    if not hashes:
+        return []
+    archives = (
+        await db.scalars(
+            select(PrintArchive)
+            .where(PrintArchive.content_hash.in_(hashes), PrintArchive.deleted_at.is_(None), *visible)
+            .order_by(PrintArchive.id)
+        )
+    ).all()
+    by_key: dict[tuple, list[PrintArchive]] = {}
+    for archive in archives:
+        by_key.setdefault(_key(archive), []).append(archive)
+    members = [a for group in by_key.values() if len(group) > 1 for a in group]
+    busy = await _busy(db, [a.id for a in members])
+    run_counts = dict(
+        (
+            await db.execute(
+                select(PrintLogEntry.archive_id, func.count(PrintLogEntry.id))
+                .where(PrintLogEntry.archive_id.in_([a.id for a in members]))
+                .group_by(PrintLogEntry.archive_id)
+            )
+        ).all()
+    )
+    groups = []
+    for group in by_key.values():
+        if len(group) < 2:
+            continue
+        skipped = None
+        if any(a.id in busy for a in group):
+            skipped = SKIPPED_BUSY
+        elif await asyncio.to_thread(_conflicting_extras, group):
+            skipped = SKIPPED_CONFLICTING_FILES
+        groups.append(
+            MergeGroup(
+                survivor=group[0],
+                merged=group[1:],
+                run_counts={a.id: run_counts.get(a.id, 0) for a in group},
+                skipped=skipped,
+            )
+        )
+    return sorted(groups, key=lambda g: g.survivor.id)
+
+
+def _digest(relpath: str) -> str | None:
+    path = _under_base(relpath)
+    if path is None or not path.is_file():
+        return None
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1 << 20), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _conflicting_extras(group: list[PrintArchive]) -> bool:
+    """Whether two archives of the group carry different source 3MFs or
+    design files: the archive keeps one, and the other is the user's."""
+    for attr in _FILE_EXTRAS:
+        digests = {_digest(getattr(a, attr)) for a in group if getattr(a, attr)}
+        if len(digests - {None}) > 1:
+            return True
+    return False
+
+
+async def _busy(db: AsyncSession, archive_ids: list[int]) -> set[int]:
+    """The archives at work (see the module docstring)."""
+    from backend.app.main import _active_prints, _expected_prints
+    from backend.app.models.pipeline_run import PipelineRun
+    from backend.app.services import timelapse_jobs
+
+    if not archive_ids:
+        return set()
+    recent = utcnow_naive() - _JUST_PRINTED
+    queries = [
+        select(PrintArchive.id).where(
+            PrintArchive.id.in_(archive_ids),
+            or_(PrintArchive.status == "printing", PrintArchive.completed_at >= recent),
+        ),
+        select(PrintLogEntry.archive_id)
+        .where(
+            PrintLogEntry.archive_id.in_(archive_ids),
+            or_(PrintLogEntry.status == RUN_OPEN, PrintLogEntry.completed_at >= recent),
+        )
+        .execution_options(**{INCLUDE_OPEN_RUNS: True}),
+        select(PrintQueueItem.archive_id).where(
+            PrintQueueItem.archive_id.in_(archive_ids), PrintQueueItem.status.in_(("pending", "printing"))
+        ),
+        select(ActivePrintSpoolman.archive_id).where(ActivePrintSpoolman.archive_id.in_(archive_ids)),
+        select(PipelineRun.source_archive_id).where(
+            PipelineRun.source_archive_id.in_(archive_ids), PipelineRun.status.in_(_PIPELINE_ACTIVE)
+        ),
+    ]
+    busy: set[int | None] = set()
+    for query in queries:
+        busy |= set((await db.scalars(query)).all())
+    # Held in memory by id: a print waiting for its start or end, work in
+    # this process (a dispatch, a conversion), an editor job.
+    in_memory = set(_active_prints.values()) | set(_expected_prints.values())
+    for archive_id in archive_ids:
+        job = timelapse_jobs.get_job(archive_id)
+        if (
+            archive_id in in_memory
+            or archive_work.is_held(archive_id)
+            or (job is not None and job.status == timelapse_jobs.RUNNING)
+        ):
+            busy.add(archive_id)
+    return {archive_id for archive_id in busy if archive_id is not None}
+
+
+# Rows that point at an archive and move with it. Wallet charges and budget
+# reservations stay on the merged one (soft-deleted, not removed): they are
+# money already booked, and the old charges are unique per archive.
+def _moved_references():
+    from backend.app.models.library import LibraryFolder
+    from backend.app.models.notification import TelegramPendingVerdict
+    from backend.app.models.pending_upload import PendingUpload
+    from backend.app.models.pipeline_run import PipelineRun
+    from backend.app.models.print_batch import PrintBatch
+    from backend.app.models.project_bom import ProjectBOMItem
+    from backend.app.models.spool_usage_history import SpoolUsageHistory
+
+    return (
+        (PrintLogEntry, "archive_id"),
+        (PrintQueueItem, "archive_id"),
+        (PrintBatch, "archive_id"),
+        (ProjectBOMItem, "archive_id"),
+        (SpoolUsageHistory, "archive_id"),
+        (TelegramPendingVerdict, "archive_id"),
+        (LibraryFolder, "archive_id"),
+        (PendingUpload, "archived_id"),
+        (PipelineRun, "source_archive_id"),
+    )
+
+
+class _Files:
+    """Files placed in the survivor's folder, removed again if the merge fails."""
+
+    def __init__(self) -> None:
+        self.created: list[Path] = []
+
+    def place(self, source: Path | None, target_dir: Path) -> Path | None:
+        """``source`` in ``target_dir`` under a free name (a link where it can)."""
+        if source is None or not source.is_file():
+            return None
+        target_dir.mkdir(parents=True, exist_ok=True)
+        target = target_dir / source.name
+        n = 2
+        while target.exists():
+            if target.resolve() == source.resolve():
+                return target
+            target = target_dir / f"{source.stem}_{n}{source.suffix}"
+            n += 1
+        try:
+            os.link(source, target)
+        except OSError:
+            try:
+                shutil.copy2(source, target)
+            except BaseException:
+                target.unlink(missing_ok=True)  # a partial copy
+                raise
+        self.created.append(target)
+        return target
+
+    def undo(self) -> None:
+        for path in self.created:
+            path.unlink(missing_ok=True)
+
+
+def _relpath(path: Path) -> str:
+    return str(path.relative_to(settings.base_dir))
+
+
+def _under_base(relpath: str | None) -> Path | None:
+    if not relpath:
+        return None
+    try:
+        return safe_join_under(settings.base_dir, relpath, http=False)
+    except PathTraversalError:
+        return None
+
+
+def _missing(relpath: str | None) -> bool:
+    path = _under_base(relpath)
+    return path is None or not path.is_file()
+
+
+def _gets_a_run(archive: PrintArchive, runs: list[PrintLogEntry]) -> bool:
+    """Whether the merge writes a run for the archive's print (``_run_from_archive``)."""
+    return not runs and archive.status in _ENDED and bool(archive.started_at or archive.completed_at)
+
+
+def _orphan_video(archive: PrintArchive, runs: list[PrintLogEntry]) -> bool:
+    """Whether the archive's video is its alone, with no print to take it."""
+    from backend.app.services.print_log import owns_media
+
+    if not archive.timelapse_path or _gets_a_run(archive, runs):
+        return False
+    if any(r.timelapse_path == archive.timelapse_path for r in runs):
+        return False
+    return not any(r.timelapse_path is None and not owns_media(r) for r in runs)
+
+
+def _latest(runs, *, where) -> PrintLogEntry | None:
+    """The latest by start of ``runs`` that ``where`` admits."""
+    return max(
+        (r for r in runs if where(r)),
+        key=lambda r: (r.started_at is not None, r.started_at or datetime.min, r.id or 0),
+        default=None,
+    )
+
+
+def _run_from_archive(archive: PrintArchive, survivor_id: int, printer_name: str | None) -> PrintLogEntry:
+    """The run of a print logged before runs were: what its archive says of it."""
+    duration = None
+    if archive.started_at and archive.completed_at and archive.completed_at > archive.started_at:
+        duration = int((archive.completed_at - archive.started_at).total_seconds())
+    return PrintLogEntry(
+        archive_id=survivor_id,
+        print_name=archive.print_name,
+        plate_id=archive.plate_id,
+        printer_id=archive.printer_id,
+        printer_name=printer_name,
+        status=archive.status,
+        started_at=archive.started_at,
+        completed_at=archive.completed_at,
+        duration_seconds=duration,
+        filament_type=archive.filament_type,
+        filament_color=archive.filament_color,
+        filament_used_grams=archive.filament_used_grams,
+        cost=archive.cost,
+        energy_kwh=archive.energy_kwh,
+        energy_cost=archive.energy_cost,
+        wear_cost=archive.wear_cost,
+        failure_reason=archive.failure_reason,
+        user_verdict=archive.user_verdict,
+        user_verdict_source=archive.user_verdict_source,
+        user_verdict_at=archive.user_verdict_at,
+        thumbnail_path=archive.thumbnail_path,
+        created_by_id=archive.created_by_id,
+        quantity=archive.quantity,
+        created_at=archive.completed_at or archive.created_at,
+    )
+
+
+def _joined_tags(*tag_lists: str | None) -> str | None:
+    seen: list[str] = []
+    for tags in tag_lists:
+        for tag in (tags or "").split(","):
+            if (tag := tag.strip()) and tag.lower() not in {t.lower() for t in seen}:
+                seen.append(tag)
+    return ", ".join(seen) or None
+
+
+def _joined_notes(*notes: str | None) -> str | None:
+    kept: list[str] = []
+    for note in notes:
+        if note and note.strip() and note.strip() not in kept:
+            kept.append(note.strip())
+    return "\n\n".join(kept) or None
+
+
+@dataclass
+class _Plan:
+    """The files of one merged archive and where they go."""
+
+    archive: PrintArchive
+    photos: dict[str, Path] = field(default_factory=dict)  # name -> source
+    videos: dict[str, Path] = field(default_factory=dict)  # relpath -> source
+    extras: dict[str, Path] = field(default_factory=dict)  # attribute -> source
+
+
+async def merge_group(db: AsyncSession, survivor_id: int, merged_ids: list[int]) -> str | None:
+    """Merge ``merged_ids`` into ``survivor_id`` and commit. Returns why it
+    was left out instead (a ``SKIPPED_*``), or None once merged."""
+    with archive_work.merging([survivor_id, *merged_ids]) as marked:
+        if not marked:
+            return SKIPPED_BUSY
+        return await _merge_marked(db, survivor_id, merged_ids)
+
+
+async def _merge_marked(db: AsyncSession, survivor_id: int, merged_ids: list[int]) -> str | None:
+    """``_merge``, rolled back (with the files it placed) on any failure."""
+    files = _Files()
+    try:
+        reason, leftover = await _merge(db, survivor_id, merged_ids, files)
+        if reason is not None:
+            # Left out before anything was written.
+            return reason
+    except BaseException:
+        await db.rollback()
+        await asyncio.to_thread(files.undo)
+        raise
+    for directory in leftover:
+        await asyncio.to_thread(shutil.rmtree, directory, True)
+    logger.info("Merged archives %s into archive %s", merged_ids, survivor_id)
+    return None
+
+
+async def _merge(
+    db: AsyncSession, survivor_id: int, merged_ids: list[int], files: _Files
+) -> tuple[str | None, list[Path]]:
+    """Merge and commit; returns why it was left out instead, or the folders
+    to remove now that it has committed."""
+    from backend.app.models.printer import Printer
+    from backend.app.services.archive import ArchiveService
+    from backend.app.services.print_log import mirror_latest_run, newest_run, owns_media
+
+    ids = [survivor_id, *merged_ids]
+    # As they are now, not as the preview or the group search found them.
+    rows = {
+        a.id: a
+        for a in await db.scalars(
+            select(PrintArchive).where(PrintArchive.id.in_(ids)).execution_options(populate_existing=True)
+        )
+    }
+    if len(rows) != len(ids) or any(a.deleted_at is not None for a in rows.values()):
+        return SKIPPED_BUSY, []
+    survivor, merged = rows[survivor_id], [rows[i] for i in merged_ids]
+    if len({_key(a) for a in rows.values()}) != 1 or await _busy(db, ids):
+        return SKIPPED_BUSY, []
+    if await asyncio.to_thread(_conflicting_extras, [survivor, *merged]):
+        return SKIPPED_CONFLICTING_FILES, []
+    runs: dict[int, list[PrintLogEntry]] = {a.id: [] for a in rows.values()}
+    for run in await db.scalars(
+        select(PrintLogEntry)
+        .where(PrintLogEntry.archive_id.in_(ids))
+        .execution_options(populate_existing=True, **{INCLUDE_OPEN_RUNS: True})
+    ):
+        if run.archive_id is not None:
+            runs[run.archive_id].append(run)
+
+    # A video on an archive alone that no print of it can take (attached by
+    # hand to one never printed): the archive keeps it only if it is the one
+    # video of the group, or it would be lost or hide another.
+    orphans = {a.id: a.timelapse_path for a in rows.values() if _orphan_video(a, runs[a.id])}
+    if orphans:
+        videos = {r.timelapse_path for own in runs.values() for r in own if r.timelapse_path} | {
+            a.timelapse_path for a in rows.values() if a.timelapse_path
+        }
+        if len(orphans) > 1 or videos - set(orphans.values()):
+            return SKIPPED_CONFLICTING_FILES, []
+
+    # Plan, reading only: the files to place in the survivor's folder. The
+    # survivor's own files where it lost them, from the first archive that
+    # still has them.
+    plans = []
+    taken: set[str] = set()
+    for archive in merged:
+        plan = _Plan(archive)
+        for name in archive.photos or []:
+            if (source := find_archive_photo(archive, name)) is not None:
+                plan.photos[name] = source
+        for relpath in {r.timelapse_path for r in runs[archive.id] if r.timelapse_path} | (
+            {archive.timelapse_path} if archive.timelapse_path else set()
+        ):
+            if (source := _under_base(relpath)) is not None:
+                plan.videos[relpath] = source
+        for attr in (*_FILE_EXTRAS, "file_path", "thumbnail_path"):
+            source = _under_base(getattr(archive, attr))
+            if attr not in taken and _missing(getattr(survivor, attr)) and source is not None and source.is_file():
+                plan.extras[attr] = source
+                taken.add(attr)
+        plans.append(plan)
+
+    # Place them off the event loop, before any write: SQLite would hold its
+    # write lock through a copy that cannot link.
+    survivor_dir = archive_dir(survivor)
+    photos_dir = archive_photos_dir(survivor)
+
+    def place(source: Path, target_dir: Path) -> Path | None:
+        return files.place(source, target_dir)
+
+    def place_all() -> list[tuple[dict[str, str], dict[str, str], dict[str, str]]]:
+        placed = []
+        for plan in plans:
+            photos = {n: p.name for n, s in plan.photos.items() if (p := place(s, photos_dir)) is not None}
+            videos = {r: _relpath(p) for r, s in plan.videos.items() if (p := place(s, survivor_dir)) is not None}
+            extras = {a: _relpath(p) for a, s in plan.extras.items() if (p := place(s, survivor_dir)) is not None}
+            placed.append((photos, videos, extras))
+        return placed
+
+    placed = await asyncio.to_thread(place_all)
+
+    printer_ids = {a.printer_id for a in rows.values() if a.printer_id is not None}
+    printer_names = dict(
+        (await db.execute(select(Printer.id, Printer.name).where(Printer.id.in_(printer_ids)))).tuples().all()
+    )
+    for archive in [survivor, *merged]:
+        own = runs[archive.id]
+        # A print logged before runs keeps its only record on its archive.
+        if _gets_a_run(archive, own):
+            name = printer_names.get(archive.printer_id) if archive.printer_id is not None else None
+            run = _run_from_archive(archive, archive.id, name)
+            run.photos = list(archive.photos or [])
+            run.timelapse_path = archive.timelapse_path
+            db.add(run)
+            own.append(run)
+        # Its video, kept on the archive alone by prints logged before
+        # runs kept their own: its latest such print's.
+        elif (
+            archive.timelapse_path
+            and archive.id not in orphans
+            and not any(r.timelapse_path == archive.timelapse_path for r in own)
+        ):
+            if (heir := _latest(own, where=lambda r: r.timelapse_path is None and not owns_media(r))) is not None:
+                heir.timelapse_path = archive.timelapse_path
+        # A one-tap link of the archive's question still answers it: on
+        # its latest print, where questions are asked now.
+        if archive.confirm_token and archive.confirm_token_used_at is None and archive.user_verdict is None:
+            token, archive.confirm_token = archive.confirm_token, None
+            await db.flush()
+            if (heir := _latest(own, where=lambda r: r.confirm_token is None)) is not None:
+                heir.confirm_token = token
+                heir.confirm_requested = True
+
+    survivor_photos = list(survivor.photos or [])
+    for plan, (photos, videos, extras) in zip(plans, placed, strict=True):
+        archive = plan.archive
+        for name in archive.photos or []:
+            if name in photos and photos[name] not in survivor_photos:
+                survivor_photos.append(photos[name])
+        for attr, relpath in extras.items():
+            setattr(survivor, attr, relpath)
+        for run in runs[archive.id]:
+            if run.photos:
+                run.photos = [photos[p] for p in run.photos if p in photos]
+            if run.timelapse_path:
+                run.timelapse_path = videos.get(run.timelapse_path)
+            # Its thumbnail was its archive's copy of the file's.
+            if run.thumbnail_path == archive.thumbnail_path:
+                run.thumbnail_path = survivor.thumbnail_path
+            run.archive_id = survivor.id
+
+        # The file's description: one note, every tag, a favourite if any.
+        survivor.notes = _joined_notes(survivor.notes, archive.notes)
+        survivor.tags = _joined_tags(survivor.tags, archive.tags)
+        survivor.is_favorite = bool(survivor.is_favorite or archive.is_favorite)
+        for attr in ("library_file_id", "external_url", "makerworld_url", "designer"):
+            if getattr(survivor, attr) is None and getattr(archive, attr) is not None:
+                setattr(survivor, attr, getattr(archive, attr))
+    await db.flush()
+
+    for archive in merged:
+        for model, column in _moved_references():
+            await db.execute(
+                update(model)
+                .where(getattr(model, column) == archive.id)
+                .values({column: survivor.id})
+                .execution_options(**{INCLUDE_OPEN_RUNS: True})
+            )
+        await db.execute(
+            update(ArchiveAlias).where(ArchiveAlias.archive_id == archive.id).values(archive_id=survivor.id)
+        )
+        db.add(ArchiveAlias(merged_archive_id=archive.id, archive_id=survivor.id))
+    survivor.photos = survivor_photos or None
+    await db.flush()
+
+    # The archive shows its latest run, its video too.
+    await mirror_latest_run(db, survivor.id)
+    newest = await newest_run(db, survivor.id)
+    if orphans:
+        # The group's one video, which no print holds.
+        [(holder_id, orphan)] = orphans.items()
+        if holder_id != survivor.id and orphan is not None:
+            orphan = placed[merged_ids.index(holder_id)][1].get(orphan)
+        survivor.timelapse_path = orphan
+        survivor.timelapse_plate_id = None
+    elif newest is not None and owns_media(newest):
+        survivor.timelapse_path = newest.timelapse_path
+        survivor.timelapse_plate_id = newest.plate_id if newest.timelapse_path else None
+
+    # Their folders go once committed, unless another archive still uses
+    # one (archived in the same second under the same name).
+    service = ArchiveService(db)
+    leftover = []
+    for archive in merged:
+        for directory in service._resolve_archive_dirs_for_delete(archive):
+            if not await _folder_in_use(db, directory, merged_ids=merged_ids):
+                leftover.append(directory)
+    for archive in merged:
+        archive.deleted_at = datetime.now(timezone.utc)
+        # Its files are the survivor's now: a later purge of the row must
+        # find nothing of theirs to remove.
+        archive.file_path = ""
+        archive.thumbnail_path = archive.timelapse_path = None
+        archive.source_3mf_path = archive.f3d_path = None
+        archive.photos = None
+    await db.commit()
+    return None, leftover
+
+
+async def _folder_in_use(db: AsyncSession, directory: Path, *, merged_ids: list[int]) -> bool:
+    """Whether ``directory`` holds a file of an archive other than the merged
+    ones (the survivor's included). Compared as paths, not as text: stored
+    paths use the platform's separator."""
+    folder = directory.resolve()
+    rows = await db.execute(
+        select(PrintArchive.file_path, PrintArchive.source_3mf_path, PrintArchive.f3d_path).where(
+            PrintArchive.id.not_in(merged_ids),
+            PrintArchive.deleted_at.is_(None),
+            or_(
+                PrintArchive.file_path.contains(directory.name, autoescape=True),
+                PrintArchive.source_3mf_path.contains(directory.name, autoescape=True),
+                PrintArchive.f3d_path.contains(directory.name, autoescape=True),
+            ),
+        )
+    )
+    for row in rows.all():
+        for relpath in row:
+            if (path := _under_base(relpath)) is not None and path.resolve().is_relative_to(folder):
+                return True
+    return False
+
+
+async def merge(db: AsyncSession, survivor_ids: list[int], visible: list) -> tuple[list[dict], list[dict]]:
+    """Merge the groups of ``survivor_ids`` as they stand now (each its
+    group's oldest archive among those ``visible`` admits). Returns what was
+    merged and what was left out, and why; one group failing leaves the
+    others as they went."""
+    # Read up front: a group that fails rolls back, which expires the rows.
+    groups = {g.survivor.id: (g.skipped, [a.id for a in g.merged]) for g in await find_groups(db, visible)}
+    done, skipped = [], []
+    for survivor_id in survivor_ids:
+        if survivor_id not in groups:
+            skipped.append({"survivor_id": survivor_id, "reason": "not_a_group"})
+            continue
+        reason, merged_ids = groups[survivor_id]
+        if reason is not None:
+            skipped.append({"survivor_id": survivor_id, "reason": reason})
+            continue
+        try:
+            reason = await merge_group(db, survivor_id, merged_ids)
+        except Exception:
+            logger.exception("Merging archives %s into archive %s failed", merged_ids, survivor_id)
+            reason = SKIPPED_FAILED
+        if reason is None:
+            done.append({"survivor_id": survivor_id, "merged_ids": merged_ids})
+        else:
+            skipped.append({"survivor_id": survivor_id, "reason": reason})
+    return done, skipped
+
+
+def merge_scope_clauses(printer_scope: PrinterScope) -> list:
+    """Only archives every print of which the caller's printers made: a
+    merge changes and removes whole archives (archive redesign, step 8)."""
+    from backend.app.core.row_visibility import whole_archive_scope_clause
+
+    clause = whole_archive_scope_clause(printer_scope)
+    return [clause] if clause is not None else []

backend/app/services/archive_work.py

@@ -0,0 +1,78 @@
+"""Archives that work running in this process holds by id (archive redesign P2).
+
+A merge moves an archive's runs, files and references to another and removes
+its folder. Work that holds the archive by its id in memory, and not in the
+database, must not have that happen under it: a queue dispatch between
+choosing the archive and recording it on its queue item, a timelapse
+conversion. Each holds the archive here for as long as it runs; a merge
+leaves a held archive out, and marks the archives it merges so a dispatch
+leaves them out in turn. Both happen on the one event loop with no await
+between the check and the mark, so they never pass each other.
+"""
+
+from __future__ import annotations
+
+from collections import Counter
+from collections.abc import Iterable, Iterator
+from contextlib import contextmanager
+
+_held: Counter[int] = Counter()
+_merging: set[int] = set()
+
+
+def hold(archive_id: int) -> None:
+    _held[archive_id] += 1
+
+
+def release(archive_id: int) -> None:
+    if _held[archive_id] <= 1:
+        _held.pop(archive_id, None)
+    else:
+        _held[archive_id] -= 1
+
+
+@contextmanager
+def held(archive_id: int) -> Iterator[None]:
+    hold(archive_id)
+    try:
+        yield
+    finally:
+        release(archive_id)
+
+
+def is_held(archive_id: int) -> bool:
+    return archive_id in _held
+
+
+def is_merging(archive_id: int) -> bool:
+    return archive_id in _merging
+
+
+@contextmanager
+def merging(archive_ids: Iterable[int]) -> Iterator[bool]:
+    """Mark ``archive_ids`` as being merged; yields False (and marks nothing)
+    when one is held or already being merged."""
+    ids = set(archive_ids)
+    if any(i in _held or i in _merging for i in ids):
+        yield False
+        return
+    _merging.update(ids)
+    try:
+        yield True
+    finally:
+        _merging.difference_update(ids)
+
+
+def held_while(archive_id: int, work):
+    """``work`` (a coroutine) with the archive held until it ends. Held now,
+    before the task it is spawned as starts, so no merge between the spawn
+    and its first step can take the archive."""
+    hold(archive_id)
+    return _release_after(archive_id, work)
+
+
+async def _release_after(archive_id: int, work):
+    try:
+        return await work
+    finally:
+        release(archive_id)

backend/app/services/print_scheduler.py

@@ -1378,6 +1378,10 @@ class PrintScheduler:
         # event-loop thread, so this dict needs no lock.
         # item_id -> (task, printer_id)
         self._inflight: dict[int, tuple[asyncio.Task, int | None]] = {}
+        # The archive each dispatch in flight prints into, held in
+        # services/archive_work so a merge leaves it alone until the dispatch
+        # ends: item_id -> archive_id. Same threading as above.
+        self._held_archives: dict[int, int] = {}
         # Expected prints registered by `_start_print` that have not yet had a
         # print command sent. Populated at registration, dropped once
         # `start_print()` succeeds, and rolled back by `_dispatch_one` on every
@@ -2847,6 +2851,8 @@ class PrintScheduler:
                 # Likewise the print user it set: the printer's next print is
                 # not this item's.
                 self._rollback_unconfirmed_print_user(item_id)
+                # Its archive is no longer held for it (archive_work).
+                self._release_held_archive(item_id)
                 # Mirror the pre-#1625 background-dispatch lifecycle: a
                 # reservation survives only after start_print() accepted the
                 # command. Failure, cancellation, deferral, and exceptions all
@@ -8054,6 +8060,20 @@ class PrintScheduler:
 
         return prev_item.status in ("completed", "cancelled")
 
+    def _hold_archive(self, item_id: int, archive_id: int) -> None:
+        """Hold the archive a dispatch prints into until it ends (archive_work)."""
+        from backend.app.services import archive_work
+
+        self._release_held_archive(item_id)
+        archive_work.hold(archive_id)
+        self._held_archives[item_id] = archive_id
+
+    def _release_held_archive(self, item_id: int) -> None:
+        from backend.app.services import archive_work
+
+        if (archive_id := self._held_archives.pop(item_id, None)) is not None:
+            archive_work.release(archive_id)
+
     @staticmethod
     async def _printing_elsewhere(db: AsyncSession, archive_id: int) -> bool:
         """Whether a print of the archive is running (its run open)."""
@@ -8107,8 +8127,15 @@ class PrintScheduler:
                 .order_by(PrintArchive.id.desc())
             )
         ).all()
+        from backend.app.services import archive_work
+
         for candidate in candidates:
+            # Not one a merge is moving into another archive right now; held
+            # from here, with no await between, until the dispatch ends.
+            if archive_work.is_merging(candidate.id):
+                continue
             if candidate.file_path and (settings.base_dir / candidate.file_path).is_file():
+                self._hold_archive(item.id, candidate.id)
                 return candidate
         return None
 
@@ -8992,6 +9019,15 @@ class PrintScheduler:
             # Print from archive
             result = await db.execute(select(PrintArchive).where(PrintArchive.id == item.archive_id))
             archive = result.scalar_one_or_none()
+            if archive is not None and archive.deleted_at is not None:
+                # Merged into another archive since it was queued (archive
+                # redesign P2): print that one, whose file it now holds.
+                from backend.app.services.archive_merge import resolve_alias
+
+                if (into := await resolve_alias(db, archive.id)) is not None:
+                    archive = await db.get(PrintArchive, into)
+                    if archive is not None:
+                        item.archive_id = archive.id
             if not archive:
                 if not await self._fail_dispatch(db, item, "Archive not found"):
                     return
@@ -9093,6 +9129,7 @@ class PrintScheduler:
                         plate_id=item.plate_id,  # selected plate → Print History (#2603)
                     )
                 if archive:
+                    self._hold_archive(item.id, archive.id)
                     item.archive_id = archive.id
                     if item.confirm_outcome:
                         archive.confirm_requested = True  # ask-for-outcome opt-in (#1898)

backend/tests/conftest.py

@@ -213,6 +213,21 @@ def mfa_encryption_isolation(monkeypatch, tmp_path):
     enc_mod._key_source = None
 
 
+@pytest.fixture(autouse=True)
+def reset_archive_work():
+    """Drop archives held or being merged between tests (services/archive_work).
+
+    Every test database hands out the same archive ids, and a dispatch driven
+    without ``_dispatch_one`` never releases its hold."""
+    from backend.app.services import archive_work
+
+    archive_work._held.clear()
+    archive_work._merging.clear()
+    yield
+    archive_work._held.clear()
+    archive_work._merging.clear()
+
+
 @pytest.fixture(autouse=True)
 def reset_printer_file_cache():
     """A fresh file-browser preview cache per test.

backend/tests/integration/test_archive_merge.py

@@ -0,0 +1,604 @@
+"""Duplicate archives of one file merge into one (archive redesign P2).
+
+Same file, owner and project: the oldest archive stays and takes the others'
+runs, references, photos and videos; the others stay soft-deleted with an
+alias, their money untouched.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.core.config import settings as app_settings
+from backend.app.models.archive import PrintArchive
+from backend.app.models.archive_alias import ArchiveAlias
+from backend.app.models.finance import WalletTransaction
+from backend.app.models.print_log import RUN_OPEN, PrintLogEntry
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.project import Project
+from backend.app.models.project_bom import ProjectBOMItem
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+HASH = "a" * 64
+T1 = datetime(2026, 9, 1, 9)
+
+
+@pytest.fixture
+def files(tmp_path, monkeypatch):
+    monkeypatch.setattr(app_settings, "base_dir", tmp_path)
+    monkeypatch.setattr(app_settings, "archive_dir", tmp_path / "archive")
+    return tmp_path
+
+
+def _dir(files, name):
+    path = files / "archive" / "1" / name
+    (path / "photos").mkdir(parents=True, exist_ok=True)
+    (path / "file.3mf").write_bytes(b"3mf")
+    return path
+
+
+@pytest.fixture
+async def dupes(db_session, printer_factory, files):
+    """X and Y printed once each, Z printed before runs were logged; W is the
+    same file in another project, V the same file deleted."""
+    printer = await printer_factory(name="A")
+    project = Project(name="Other")
+    db_session.add(project)
+    await db_session.flush()
+
+    def archive(name, **fields):
+        folder = _dir(files, name)
+        row = PrintArchive(
+            printer_id=printer.id,
+            filename="part.3mf",
+            print_name="part",
+            file_path=str((folder / "file.3mf").relative_to(files)),
+            file_size=3,
+            content_hash=HASH,
+            thumbnail_path=f"archive/1/{name}/thumb.png",
+            **({"status": "completed"} | fields),
+        )
+        db_session.add(row)
+        return row, folder
+
+    x, x_dir = archive("x", notes="first", tags="red", started_at=T1, completed_at=T1 + timedelta(hours=1))
+    y, y_dir = archive(
+        "y",
+        notes="second",
+        tags="blue, red",
+        is_favorite=True,
+        started_at=T1 + timedelta(days=1),
+        completed_at=T1 + timedelta(days=1, hours=1),
+        confirm_requested=True,
+        confirm_token="tok-y",
+        photos=["finish_y.jpg"],
+        timelapse_path="archive/1/y/y.mp4",
+    )
+    z, z_dir = archive(
+        "z",
+        status="failed",
+        failure_reason="spaghetti",
+        started_at=T1 + timedelta(days=2),
+        completed_at=T1 + timedelta(days=2, hours=1),
+        photos=["finish_z.jpg"],
+        timelapse_path="archive/1/z/z.mp4",
+        filament_used_grams=12.0,
+    )
+    w, _ = archive("w", project_id=project.id)
+    v, _ = archive("v", deleted_at=datetime(2026, 10, 1))
+    await db_session.flush()
+    (y_dir / "photos" / "finish_y.jpg").write_bytes(b"y")
+    (y_dir / "y.mp4").write_bytes(b"yy")
+    (z_dir / "photos" / "finish_z.jpg").write_bytes(b"z")
+    (z_dir / "z.mp4").write_bytes(b"zz")
+
+    r1 = PrintLogEntry(
+        archive_id=x.id, printer_id=printer.id, status="completed", started_at=T1, completed_at=T1 + timedelta(hours=1)
+    )
+    r2 = PrintLogEntry(
+        archive_id=y.id,
+        printer_id=printer.id,
+        status="completed",
+        started_at=T1 + timedelta(days=1),
+        completed_at=T1 + timedelta(days=1, hours=1),
+        photos=["finish_y.jpg"],
+        timelapse_path="archive/1/y/y.mp4",
+        thumbnail_path="archive/1/y/thumb.png",
+    )
+    item = PrintQueueItem(printer_id=printer.id, archive_id=y.id, status="completed", position=1)
+    bom = ProjectBOMItem(project_id=project.id, name="part", archive_id=y.id)
+    charge = WalletTransaction(transaction_type="print_charge", amount=-1.0, print_archive_id=y.id)
+    db_session.add_all([r1, r2, item, bom, charge])
+    await db_session.commit()
+    return SimpleNamespace(
+        x=x.id,
+        y=y.id,
+        z=z.id,
+        w=w.id,
+        v=v.id,
+        r1=r1.id,
+        r2=r2.id,
+        item=item.id,
+        bom=bom.id,
+        charge=charge.id,
+        x_dir=x_dir,
+        y_dir=y_dir,
+        z_dir=z_dir,
+        printer=printer.id,
+    )
+
+
+async def _row(db_session, model, row_id):
+    db_session.expire_all()
+    return await db_session.get(model, row_id)
+
+
+async def _runs(db_session, archive_id):
+    db_session.expire_all()
+    return list(
+        await db_session.scalars(
+            select(PrintLogEntry).where(PrintLogEntry.archive_id == archive_id).order_by(PrintLogEntry.started_at)
+        )
+    )
+
+
+async def test_the_preview_shows_the_group(async_client, dupes):
+    response = await async_client.get("/api/v1/archives/merge-preview")
+
+    assert response.status_code == 200, response.text
+    [group] = response.json()["groups"]
+    assert group["survivor"]["id"] == dupes.x
+    assert [(m["id"], m["run_count"]) for m in group["merged"]] == [(dupes.y, 1), (dupes.z, 0)]
+    assert group["skipped"] is None
+
+
+async def test_merging_keeps_the_oldest_with_every_print(async_client, db_session, dupes):
+    response = await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    assert response.status_code == 200, response.text
+    assert response.json() == {"merged": [{"survivor_id": dupes.x, "merged_ids": [dupes.y, dupes.z]}], "skipped": []}
+
+    runs = await _runs(db_session, dupes.x)
+    assert [r.status for r in runs] == ["completed", "completed", "failed"]
+    from_z = runs[2]
+    assert (from_z.failure_reason, from_z.filament_used_grams, from_z.photos) == ("spaghetti", 12.0, ["finish_z.jpg"])
+    archive = await _row(db_session, PrintArchive, dupes.x)
+    # The file's description, and its latest print.
+    assert (archive.notes, archive.tags, archive.is_favorite) == ("first\n\nsecond", "red, blue", True)
+    assert (archive.status, archive.failure_reason) == ("failed", "spaghetti")
+    assert archive.photos == ["finish_y.jpg", "finish_z.jpg"]
+
+
+async def test_the_others_stay_deleted_with_their_money(async_client, db_session, dupes):
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    for merged in (dupes.y, dupes.z):
+        assert (await _row(db_session, PrintArchive, merged)).deleted_at is not None
+        assert (await db_session.get(ArchiveAlias, merged)).archive_id == dupes.x
+    assert (await _row(db_session, WalletTransaction, dupes.charge)).print_archive_id == dupes.y
+    assert (await _row(db_session, PrintArchive, dupes.w)).deleted_at is None
+
+
+async def test_their_references_move(async_client, db_session, dupes):
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    assert (await _row(db_session, PrintQueueItem, dupes.item)).archive_id == dupes.x
+    assert (await _row(db_session, ProjectBOMItem, dupes.bom)).archive_id == dupes.x
+
+
+async def test_their_photos_and_videos_move_into_its_folder(async_client, db_session, dupes):
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    assert (dupes.x_dir / "photos" / "finish_y.jpg").read_bytes() == b"y"
+    assert (dupes.x_dir / "y.mp4").read_bytes() == b"yy"
+    r2 = await _row(db_session, PrintLogEntry, dupes.r2)
+    assert (r2.timelapse_path, r2.thumbnail_path) == ("archive/1/x/y.mp4", "archive/1/x/thumb.png")
+    runs = await _runs(db_session, dupes.x)
+    assert runs[2].timelapse_path == "archive/1/x/z.mp4"
+    assert (await _row(db_session, PrintArchive, dupes.x)).timelapse_path == "archive/1/x/z.mp4"
+    assert not dupes.y_dir.exists() and not dupes.z_dir.exists()
+
+
+async def test_a_merged_id_finds_the_archive(async_client, dupes):
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    response = await async_client.get(f"/api/v1/archives/{dupes.y}")
+
+    assert response.status_code == 200, response.text
+    assert response.json()["id"] == dupes.x
+
+
+async def test_its_one_tap_link_still_answers_its_print(async_client, db_session, dupes):
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    response = await async_client.post("/api/v1/archives/confirm/tok-y/good")
+
+    assert response.status_code == 200, response.text
+    assert (await _row(db_session, PrintLogEntry, dupes.r2)).user_verdict == "good"
+
+
+async def test_a_group_at_work_waits(async_client, db_session, dupes):
+    db_session.add(PrintLogEntry(archive_id=dupes.y, printer_id=dupes.printer, status=RUN_OPEN, started_at=T1))
+    await db_session.commit()
+
+    [group] = (await async_client.get("/api/v1/archives/merge-preview")).json()["groups"]
+    assert group["skipped"] == "busy"
+    response = await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+    assert response.json() == {"merged": [], "skipped": [{"survivor_id": dupes.x, "reason": "busy"}]}
+    assert (await _row(db_session, PrintArchive, dupes.y)).deleted_at is None
+
+
+async def test_a_group_just_printed_waits(async_client, db_session, dupes):
+    """Its timelapse and finish photo may still be on their way."""
+    from backend.app.utils.local_time import utcnow_naive
+
+    run = await _row(db_session, PrintLogEntry, dupes.r2)
+    run.completed_at = utcnow_naive() - timedelta(minutes=5)
+    await db_session.commit()
+
+    [group] = (await async_client.get("/api/v1/archives/merge-preview")).json()["groups"]
+    assert group["skipped"] == "busy"
+
+
+async def test_a_queued_group_waits(async_client, db_session, dupes):
+    db_session.add(PrintQueueItem(printer_id=dupes.printer, archive_id=dupes.z, status="pending", position=2))
+    await db_session.commit()
+
+    [group] = (await async_client.get("/api/v1/archives/merge-preview")).json()["groups"]
+    assert group["skipped"] == "busy"
+
+
+async def test_a_survivor_printed_before_runs_keeps_its_print(async_client, db_session, dupes):
+    await db_session.delete(await _row(db_session, PrintLogEntry, dupes.r1))
+    await db_session.commit()
+
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    runs = await _runs(db_session, dupes.x)
+    assert [r.started_at for r in runs] == [T1, T1 + timedelta(days=1), T1 + timedelta(days=2)]
+
+
+async def test_a_failed_merge_leaves_everything(async_client, db_session, dupes):
+    with patch("backend.app.services.print_log.mirror_latest_run", side_effect=RuntimeError("boom")):
+        response = await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    assert response.json() == {"merged": [], "skipped": [{"survivor_id": dupes.x, "reason": "failed"}]}
+
+    assert (await _row(db_session, PrintArchive, dupes.y)).deleted_at is None
+    assert (await _row(db_session, PrintLogEntry, dupes.r2)).archive_id == dupes.y
+    assert not (dupes.x_dir / "y.mp4").exists()
+    assert (dupes.y_dir / "y.mp4").exists()
+
+
+async def test_merging_needs_delete_all(async_client, db_session, dupes):
+    from backend.tests.integration.test_archives_last_per_printer import _admin_token, _auth, _user
+
+    admin = await _admin_token(async_client)
+    jwt, _ = await _user(async_client, admin, "editor", ["archives:read_all", "archives:update_all"])
+
+    response = await async_client.post("/api/v1/archives/merge", headers=_auth(jwt), json={"survivor_ids": [dupes.x]})
+
+    assert response.status_code == 403
+
+
+async def test_a_scoped_user_merges_only_archives_wholly_theirs(async_client, db_session, printer_factory, dupes):
+    from backend.tests.integration.test_archives_last_per_printer import _admin_token, _auth, _user
+
+    other = await printer_factory(name="B")
+    db_session.add(PrintLogEntry(archive_id=dupes.z, printer_id=other.id, status="completed", started_at=T1))
+    await db_session.commit()
+    admin = await _admin_token(async_client)
+    jwt, _ = await _user(
+        async_client,
+        admin,
+        "scoped",
+        ["archives:read_all", "archives:update_all", "archives:delete_all"],
+        printer_ids=[dupes.printer],
+    )
+
+    [group] = (await async_client.get("/api/v1/archives/merge-preview", headers=_auth(jwt))).json()["groups"]
+
+    assert [m["id"] for m in group["merged"]] == [dupes.y]
+
+
+async def test_a_group_tracked_by_spoolman_waits(async_client, db_session, dupes):
+    from backend.app.models.active_print_spoolman import ActivePrintSpoolman
+
+    db_session.add(ActivePrintSpoolman(printer_id=dupes.printer, archive_id=dupes.y, ams_trays={}))
+    await db_session.commit()
+
+    [group] = (await async_client.get("/api/v1/archives/merge-preview")).json()["groups"]
+    assert group["skipped"] == "busy"
+
+
+async def test_a_group_printing_waits(async_client, db_session, dupes):
+    (await _row(db_session, PrintArchive, dupes.z)).status = "printing"
+    await db_session.commit()
+
+    [group] = (await async_client.get("/api/v1/archives/merge-preview")).json()["groups"]
+    assert group["skipped"] == "busy"
+
+
+async def test_an_archive_merged_later_passes_on_its_aliases(db_session, dupes):
+    """Y went into X; X later goes into O: Y's id finds O in one step."""
+    from backend.app.services.archive_merge import merge_group, resolve_alias
+
+    assert await merge_group(db_session, dupes.x, [dupes.y]) is None
+    # W takes X in, as a merge keeping W would (same file, same project now).
+    w = await _row(db_session, PrintArchive, dupes.w)
+    w.project_id = None
+    await db_session.commit()
+    assert await merge_group(db_session, dupes.w, [dupes.x]) is None
+
+    assert await resolve_alias(db_session, dupes.y) == dupes.w
+    assert await resolve_alias(db_session, dupes.x) == dupes.w
+
+
+async def test_a_video_only_its_archive_held_goes_with_its_print(async_client, db_session, dupes):
+    """Y's print was logged before runs kept their own video: Y alone held it."""
+    r2 = await _row(db_session, PrintLogEntry, dupes.r2)
+    r2.timelapse_path, r2.photos = None, None
+    await db_session.commit()
+
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    assert (await _row(db_session, PrintLogEntry, dupes.r2)).timelapse_path == "archive/1/x/y.mp4"
+
+
+async def test_the_survivors_own_old_video_stays_with_its_print(async_client, db_session, files, dupes):
+    (dupes.x_dir / "x.mp4").write_bytes(b"xx")
+    (await _row(db_session, PrintArchive, dupes.x)).timelapse_path = "archive/1/x/x.mp4"
+    await db_session.commit()
+
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    assert (await _row(db_session, PrintLogEntry, dupes.r1)).timelapse_path == "archive/1/x/x.mp4"
+
+
+async def test_the_survivors_own_one_tap_link_answers_its_own_print(async_client, db_session, dupes):
+    x = await _row(db_session, PrintArchive, dupes.x)
+    x.confirm_requested, x.confirm_token = True, "tok-x"
+    await db_session.commit()
+
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+    response = await async_client.post("/api/v1/archives/confirm/tok-x/reject")
+
+    assert response.status_code == 200, response.text
+    assert (await _row(db_session, PrintLogEntry, dupes.r1)).user_verdict == "reject"
+    assert (await _row(db_session, PrintLogEntry, dupes.r2)).user_verdict is None
+
+
+async def test_a_folder_shared_with_the_survivor_stays(async_client, db_session, files, dupes):
+    """Archived in the same second under the same name: one folder."""
+    y = await _row(db_session, PrintArchive, dupes.y)
+    y.file_path = "archive/1/x/file.3mf"
+    await db_session.commit()
+
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    assert (dupes.x_dir / "file.3mf").exists()
+    # Nothing of the survivor's for a later purge of the merged row to remove.
+    assert (await _row(db_session, PrintArchive, dupes.y)).file_path == ""
+
+
+async def test_a_folder_another_archive_uses_stays(async_client, db_session, files, dupes):
+    other = PrintArchive(filename="o.3mf", file_path="archive/1/y/other.3mf", file_size=1, created_by_id=None)
+    db_session.add(other)
+    await db_session.commit()
+
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    assert dupes.y_dir.exists()
+
+
+async def test_different_source_files_wait_for_the_user(async_client, db_session, files, dupes):
+    for archive_id, name, content in ((dupes.x, "x", b"one"), (dupes.y, "y", b"two")):
+        (files / "archive" / "1" / name / "src.3mf").write_bytes(content)
+        (await _row(db_session, PrintArchive, archive_id)).source_3mf_path = f"archive/1/{name}/src.3mf"
+        await db_session.commit()
+
+    [group] = (await async_client.get("/api/v1/archives/merge-preview")).json()["groups"]
+
+    assert group["skipped"] == "conflicting_files"
+
+
+async def test_the_same_source_file_twice_merges(async_client, db_session, files, dupes):
+    for archive_id, name in ((dupes.x, "x"), (dupes.y, "y")):
+        (files / "archive" / "1" / name / "src.3mf").write_bytes(b"same")
+        (await _row(db_session, PrintArchive, archive_id)).source_3mf_path = f"archive/1/{name}/src.3mf"
+        await db_session.commit()
+
+    [group] = (await async_client.get("/api/v1/archives/merge-preview")).json()["groups"]
+
+    assert group["skipped"] is None
+
+
+@pytest.mark.parametrize("held_by", ["dispatch", "active_print", "pipeline", "timelapse_job"])
+async def test_an_archive_held_by_work_in_flight_waits(async_client, db_session, dupes, held_by, monkeypatch):
+    from backend.app import main as main_module
+    from backend.app.services import archive_work, timelapse_jobs
+
+    if held_by == "dispatch":
+        archive_work.hold(dupes.y)
+    elif held_by == "active_print":
+        monkeypatch.setitem(main_module._active_prints, (dupes.printer, "part.3mf"), dupes.y)
+    elif held_by == "pipeline":
+        from backend.app.models.pipeline_run import PipelineRun
+
+        db_session.add(PipelineRun(source_archive_id=dupes.y, status="slicing"))
+        await db_session.commit()
+    else:
+        monkeypatch.setitem(timelapse_jobs._jobs, dupes.y, SimpleNamespace(status=timelapse_jobs.RUNNING))
+
+    [group] = (await async_client.get("/api/v1/archives/merge-preview")).json()["groups"]
+
+    assert group["skipped"] == "busy"
+
+
+async def test_a_held_archive_is_left_alone_even_after_the_preview(db_session, dupes):
+    """Held between the group search and its merge: a dispatch chose it."""
+    from backend.app.services import archive_merge, archive_work
+
+    real_find = archive_merge.find_groups
+
+    async def find_then_hold(db, visible):
+        groups = await real_find(db, visible)
+        archive_work.hold(dupes.y)
+        return groups
+
+    with patch.object(archive_merge, "find_groups", find_then_hold):
+        merged, skipped = await archive_merge.merge(db_session, [dupes.x], [])
+
+    assert (merged, skipped) == ([], [{"survivor_id": dupes.x, "reason": "busy"}])
+
+
+async def test_an_edit_made_after_the_preview_is_kept(db_session, test_engine, dupes):
+    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+    from backend.app.services.archive_merge import merge_group
+
+    x = await _row(db_session, PrintArchive, dupes.x)
+    assert x.notes == "first"
+    async with async_sessionmaker(test_engine, class_=AsyncSession)() as other:
+        (await other.get(PrintArchive, dupes.x)).notes = "edited"
+        await other.commit()
+
+    assert await merge_group(db_session, dupes.x, [dupes.y, dupes.z]) is None
+    assert (await _row(db_session, PrintArchive, dupes.x)).notes == "edited\n\nsecond"
+
+
+async def test_a_dispatch_leaves_an_archive_being_merged(db_session, dupes):
+    from backend.app.services import archive_work
+
+    with archive_work.merging([dupes.y]) as marked:
+        assert marked
+        with archive_work.merging([dupes.y]) as again:
+            assert not again
+
+
+async def test_a_group_that_got_to_work_after_the_preview_waits(db_session, dupes):
+    """Queued between the group search and its merge."""
+    from backend.app.services import archive_merge
+
+    real_find = archive_merge.find_groups
+
+    async def find_then_queue(db, visible):
+        groups = await real_find(db, visible)
+        db.add(PrintQueueItem(printer_id=dupes.printer, archive_id=dupes.z, status="pending", position=3))
+        await db.commit()
+        return groups
+
+    with patch.object(archive_merge, "find_groups", find_then_queue):
+        merged, skipped = await archive_merge.merge(db_session, [dupes.x], [])
+
+    assert (merged, skipped) == ([], [{"survivor_id": dupes.x, "reason": "busy"}])
+
+
+async def test_a_failing_group_leaves_the_next_to_merge(db_session, files, dupes):
+    from backend.app.services import archive_merge
+
+    first = PrintArchive(filename="o.3mf", file_path="archive/1/o1/file.3mf", file_size=1, content_hash="b" * 64)
+    second = PrintArchive(filename="o.3mf", file_path="archive/1/o2/file.3mf", file_size=1, content_hash="b" * 64)
+    db_session.add_all([first, second])
+    await db_session.commit()
+    first_id, second_id = first.id, second.id
+    real = archive_merge._merge
+    calls = []
+
+    async def fail_the_first(db, survivor_id, merged_ids, files_):
+        calls.append(survivor_id)
+        if len(calls) == 1:
+            await db.flush()
+            raise RuntimeError("boom")
+        return await real(db, survivor_id, merged_ids, files_)
+
+    with patch.object(archive_merge, "_merge", fail_the_first):
+        merged, skipped = await archive_merge.merge(db_session, [dupes.x, first_id], [])
+
+    assert skipped == [{"survivor_id": dupes.x, "reason": "failed"}]
+    assert merged == [{"survivor_id": first_id, "merged_ids": [second_id]}]
+
+
+async def test_a_video_attached_by_hand_to_an_unprinted_archive_stays(async_client, db_session, files, dupes):
+    """Y was uploaded, never printed, and given a video by hand: the group's
+    only video, so the archive keeps it."""
+    await db_session.delete(await _row(db_session, PrintLogEntry, dupes.r2))
+    await db_session.commit()
+    for archive_id in (dupes.y, dupes.z):
+        (await _row(db_session, PrintArchive, archive_id)).status = "archived"
+        await db_session.commit()
+    (await _row(db_session, PrintArchive, dupes.z)).timelapse_path = None
+    await db_session.commit()
+
+    response = await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+    assert response.json()["skipped"] == [], response.text
+
+    assert (await _row(db_session, PrintArchive, dupes.x)).timelapse_path == "archive/1/x/y.mp4"
+    assert (dupes.x_dir / "y.mp4").exists()
+
+
+async def test_a_video_attached_by_hand_beside_another_waits(async_client, db_session, files, dupes):
+    """Z's print brings its own video: the archive could show only one."""
+    await db_session.delete(await _row(db_session, PrintLogEntry, dupes.r2))
+    await db_session.commit()
+    (await _row(db_session, PrintArchive, dupes.y)).status = "archived"
+    await db_session.commit()
+
+    [group] = (await async_client.get("/api/v1/archives/merge-preview")).json()["groups"]
+    response = await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    assert group["skipped"] is None
+    assert response.json()["skipped"] == [{"survivor_id": dupes.x, "reason": "conflicting_files"}]
+
+
+async def test_the_survivors_lost_file_comes_from_a_duplicate(async_client, db_session, files, dupes):
+    (dupes.x_dir / "file.3mf").unlink()
+
+    await async_client.post("/api/v1/archives/merge", json={"survivor_ids": [dupes.x]})
+
+    archive = await _row(db_session, PrintArchive, dupes.x)
+    assert (files / archive.file_path).read_bytes() == b"3mf"
+
+
+async def test_a_copy_that_fails_leaves_no_partial_file(files):
+    from backend.app.services.archive_merge import _Files
+
+    source = files / "big.mp4"
+    source.write_bytes(b"x" * 100)
+    target_dir = files / "target"
+
+    def partial_copy(src, dst):
+        Path(dst).write_bytes(b"x")
+        raise OSError("disk full")
+
+    with (
+        patch("backend.app.services.archive_merge.os.link", side_effect=OSError("cross-device")),
+        patch("backend.app.services.archive_merge.shutil.copy2", side_effect=partial_copy),
+        pytest.raises(OSError),
+    ):
+        _Files().place(source, target_dir)
+
+    assert list(target_dir.iterdir()) == []
+
+
+async def test_an_avi_conversion_holds_its_archive_from_the_spawn(db_session, files, dupes):
+    from backend.app.services import archive_work
+    from backend.app.services.archive import ArchiveService
+
+    spawned = []
+    with (
+        patch("backend.app.services.archive.spawn_background_task", side_effect=lambda c, **_: spawned.append(c)),
+        patch("backend.app.services.archive._announce_timelapse"),
+    ):
+        assert await ArchiveService(db_session).attach_timelapse(dupes.x, b"avi", "v.avi")
+
+    assert archive_work.is_held(dupes.x)
+    for coro in spawned:
+        coro.close()

backend/tests/integration/test_print_complete_end_to_end.py

@@ -560,3 +560,45 @@ async def test_the_reconcile_reads_the_plate_the_queue_job_printed(test_engine,
 
     assert end_time.call_args.args[2] == 3
     assert complete.call_args.args[1]["status"] == "completed"
+
+
+async def test_the_completions_media_work_holds_the_archive(test_engine, db_session, queued_print):
+    """The scan, the layer stitch and the finish-photo upgrade find their
+    print by the archive's id: held against a merge (archive redesign P2)."""
+    from backend.app.services import archive_work
+
+    held = []
+    real = archive_work.held_while
+
+    def spy(archive_id, work):
+        held.append((archive_id, work.__name__))
+        return real(archive_id, work)
+
+    from backend.app.core.tasks import spawn_background_task
+
+    # One shared SQLite connection: tasks take turns, but for the notification.
+    turn = asyncio.Lock()
+
+    async def in_turn(coro):
+        async with turn:
+            return await coro
+
+    def spawn(coro, name=None, **kwargs):
+        return spawn_background_task(coro if name == "photo-then-notify" else in_turn(coro), name=name, **kwargs)
+
+    with (
+        patch("backend.app.main.spawn_background_task", side_effect=spawn),
+        patch("backend.app.services.camera.capture_finish_photo", AsyncMock(return_value=None)),
+        patch.object(archive_work, "held_while", side_effect=spy),
+        patch("backend.app.main._scan_for_timelapse_with_retries", AsyncMock()),
+        patch("backend.app.main._capture_finish_photo_from_timelapse", AsyncMock(return_value=(None, True))),
+        patch("backend.app.main._upgrade_finish_photo_from_timelapse", AsyncMock()),
+        patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+    ):
+        await _complete(test_engine, queued_print["printer"], "completed", extra={"timelapse_was_active": True})
+
+    assert {name for archive_id, name in held if archive_id == queued_print["archive"]} >= {
+        "_background_layer_timelapse"
+    }
+    assert sum(1 for archive_id, _ in held if archive_id == queued_print["archive"]) >= 3
+    assert not archive_work.is_held(queued_print["archive"])

backend/tests/unit/test_library_dispatch_reuses_archive.py

@@ -331,3 +331,87 @@ async def test_while_it_prints_elsewhere_its_plate_and_cost_centre_stay_that_pri
 
     assert item.archive_id == existing
     assert (archive.plate_id, archive.cost_center_id) == (1, 5)
+
+
+@pytest.mark.asyncio
+async def test_not_an_archive_being_merged(case):
+    """A merge is moving it into another archive (archive redesign P2)."""
+    from backend.app.services import archive_work
+
+    existing = await case.archive()
+    with archive_work.merging([existing]):
+        assert await _claim(case, PrintScheduler(), await case.item()) is None
+
+
+@pytest.mark.asyncio
+async def test_the_chosen_archive_is_held_until_the_dispatch_ends(case):
+    from backend.app.services import archive_work
+
+    existing = await case.archive()
+    scheduler = PrintScheduler()
+    item_id = await case.item()
+
+    assert await _claim(case, scheduler, item_id) == existing
+    assert archive_work.is_held(existing)
+    scheduler._release_held_archive(item_id)
+    assert not archive_work.is_held(existing)
+
+
+@pytest.mark.parametrize("fails", [False, True], ids=["sent", "failed"])
+@pytest.mark.asyncio
+async def test_the_dispatch_releases_its_archive_when_it_ends(fails):
+    """Sent or failed, a merge may take the archive again; another dispatch's
+    hold is left alone."""
+    from backend.app.services import archive_work
+
+    scheduler = PrintScheduler()
+
+    async def start_print(db, item):
+        scheduler._hold_archive(597, 298)
+        if fails:
+            raise RuntimeError("upload failed")
+
+    scheduler._hold_archive(600, 300)
+    db = MagicMock()
+    db.get = AsyncMock(return_value=MagicMock(id=597))
+    session = MagicMock()
+    session.__aenter__ = AsyncMock(return_value=db)
+    session.__aexit__ = AsyncMock(return_value=False)
+    with ExitStack() as stack:
+        stack.enter_context(patch("backend.app.services.print_scheduler.async_session", return_value=session))
+        stack.enter_context(patch.object(scheduler, "_claim_for_dispatch", AsyncMock(return_value=True)))
+        stack.enter_context(patch.object(scheduler, "_start_print", side_effect=start_print))
+        stack.enter_context(patch.object(scheduler, "_clear_dispatch_claim", AsyncMock()))
+        if fails:
+            stack.enter_context(pytest.raises(RuntimeError))
+        await scheduler._dispatch_one(597)
+
+    assert (archive_work.is_held(298), archive_work.is_held(300)) == (False, True)
+
+
+@pytest.mark.asyncio
+async def test_a_queued_print_of_a_merged_archive_prints_the_one_it_went_into(case):
+    """Queued while the merge ran (archive redesign P2)."""
+    from backend.app.models.archive_alias import ArchiveAlias
+
+    survivor = await case.archive()
+    merged = await case.archive(deleted_at=datetime(2026, 10, 31), tags="merged")
+    async with case.session_maker() as db:
+        db.add(ArchiveAlias(merged_archive_id=merged, archive_id=survivor))
+        await db.commit()
+
+    item, archive = await _start(case, await case.item(archive_id=merged, library_file_id=None), AsyncMock())
+
+    assert (item.archive_id, archive.id) == (survivor, survivor)
+
+
+def test_the_completions_media_work_holds_the_archive():
+    """Held from the spawn: a merge would move the print's run away."""
+    from backend.app.services import archive_work
+
+    spawned = []
+    with patch.object(main_module, "spawn_background_task", side_effect=lambda coro, **_: spawned.append(coro)):
+        main_module._schedule_timelapse_scan(41, 1, plate_id=None, run_id=7)
+
+    assert archive_work.is_held(41)
+    spawned[0].close()